From 165d1a5ed51d268ccabba6cb3f68961b779a6751 Mon Sep 17 00:00:00 2001 From: MaciejZiel Date: Wed, 14 Oct 2026 01:58:14 +0200 Subject: [PATCH 1/7] Show only the last error line and a stable worker label on the public dashboard --- src/pgqueue/admin/metrics.py | 12 +-- src/pgqueue/admin/queries.py | 59 +++++++++--- src/pgqueue/admin/redact.py | 35 ++++++++ src/pgqueue/admin/schemas.py | 5 +- src/pgqueue/admin/static/dashboard.js | 2 +- tests/test_admin_redaction.py | 125 ++++++++++++++++++++++++++ 6 files changed, 216 insertions(+), 22 deletions(-) create mode 100644 src/pgqueue/admin/redact.py create mode 100644 tests/test_admin_redaction.py diff --git a/src/pgqueue/admin/metrics.py b/src/pgqueue/admin/metrics.py index 17372de..22222b9 100644 --- a/src/pgqueue/admin/metrics.py +++ b/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", diff --git a/src/pgqueue/admin/queries.py b/src/pgqueue/admin/queries.py index 21091e5..2981296 100644 --- a/src/pgqueue/admin/queries.py +++ b/src/pgqueue/admin/queries.py @@ -13,6 +13,7 @@ from sqlalchemy import ColumnElement, and_, case, func, or_, select, text, update from sqlalchemy.ext.asyncio import AsyncConnection +from pgqueue.admin.redact import public_error, worker_label from pgqueue.admin.schemas import ( FeedItem, JobOut, @@ -41,7 +42,16 @@ FAILURE_STATUSES = (JobStatus.FAILED, JobStatus.DEAD) -async def overview(conn: AsyncConnection, *, minutes: int = 60) -> Overview: +def _redact_job(job: JobOut) -> JobOut: + return job.model_copy( + update={ + "last_error": public_error(job.last_error), + "locked_by": worker_label(job.locked_by) if job.locked_by else None, + } + ) + + +async def overview(conn: AsyncConnection, *, minutes: int = 60, redact: bool = False) -> Overview: now: datetime = (await conn.execute(select(func.now()))).scalar_one() queues, totals = await _queue_stats(conn, now) @@ -50,7 +60,7 @@ async def overview(conn: AsyncConnection, *, minutes: int = 60) -> Overview: totals=totals, queues=queues, throughput=await throughput(conn, now=now, minutes=minutes), - workers=await list_workers(conn), + workers=await list_workers(conn, redact=redact), periodic=await list_periodic(conn), ) @@ -169,12 +179,18 @@ async def throughput( return points -async def list_workers(conn: AsyncConnection) -> list[WorkerOut]: +async def list_workers(conn: AsyncConnection, *, redact: bool = False) -> list[WorkerOut]: alive = workers.c.last_seen_at >= func.now() - func.make_interval( 0, 0, 0, 0, 0, 0, workers.c.lease_seconds ) rows = await conn.execute(select(workers, alive.label("alive")).order_by(workers.c.started_at)) - return [WorkerOut.model_validate(row._mapping) for row in rows] + result = [WorkerOut.model_validate(row._mapping) for row in rows] + if redact: + result = [ + w.model_copy(update={"id": worker_label(w.id), "hostname": None, "pid": None}) + for w in result + ] + return result async def list_periodic(conn: AsyncConnection) -> list[PeriodicOut]: @@ -190,6 +206,7 @@ async def list_jobs( task: str | None = None, before_id: int | None = None, limit: int = 50, + redact: bool = False, ) -> JobPage: """Newest first, keyset-paginated on id (stable while jobs are inserted).""" conditions: list[ColumnElement[bool]] = [] @@ -207,12 +224,18 @@ async def list_jobs( ) ).all() page = [JobOut.model_validate(row._mapping) for row in rows[:limit]] - return JobPage(jobs=page, next_before_id=page[-1].id if len(rows) > limit else None) + next_before_id = page[-1].id if len(rows) > limit else None + if redact: + page = [_redact_job(job) for job in page] + return JobPage(jobs=page, next_before_id=next_before_id) -async def get_job(conn: AsyncConnection, job_id: int) -> JobOut | None: +async def get_job(conn: AsyncConnection, job_id: int, *, redact: bool = False) -> JobOut | None: row = (await conn.execute(select(jobs).where(jobs.c.id == job_id))).first() - return JobOut.model_validate(row._mapping) if row is not None else None + if row is None: + return None + job = JobOut.model_validate(row._mapping) + return _redact_job(job) if redact else job def _requeue_values() -> dict[str, object]: @@ -319,7 +342,9 @@ async def live_stats(conn: AsyncConnection, *, now: datetime, window: int = 120) ) -async def running_jobs(conn: AsyncConnection, *, limit: int = 40) -> list[RunningJob]: +async def running_jobs( + conn: AsyncConnection, *, limit: int = 40, redact: bool = False +) -> list[RunningJob]: """Jobs currently holding a lease, most recently started first.""" rows = await conn.execute( select( @@ -337,11 +362,14 @@ async def running_jobs(conn: AsyncConnection, *, limit: int = 40) -> list[Runnin .order_by(jobs.c.started_at.desc()) .limit(limit) ) - return [RunningJob.model_validate(row._mapping) for row in rows] + running = [RunningJob.model_validate(row._mapping) for row in rows] + if redact: + running = [j.model_copy(update={"locked_by": worker_label(j.locked_by)}) for j in running] + return running async def recent_activity( - conn: AsyncConnection, *, limit: int = 25, lookback_ids: int = 400 + conn: AsyncConnection, *, limit: int = 25, lookback_ids: int = 400, redact: bool = False ) -> list[FeedItem]: """The most recently changed jobs among the newest ``lookback_ids`` ones. @@ -372,7 +400,10 @@ async def recent_activity( ) items = [] for row in rows: - error = row.last_error.strip().splitlines()[-1] if row.last_error else None + if redact: + error = public_error(row.last_error) + else: + error = row.last_error.strip().splitlines()[-1] if row.last_error else None items.append( FeedItem( id=row.id, @@ -401,9 +432,9 @@ async def snapshot( read_only=read_only, totals=totals, queues=queues, - workers=await list_workers(conn), - running=await running_jobs(conn), + workers=await list_workers(conn, redact=read_only), + running=await running_jobs(conn, redact=read_only), periodic=await list_periodic(conn), stats=await live_stats(conn, now=now), - feed=await recent_activity(conn), + feed=await recent_activity(conn, redact=read_only), ) diff --git a/src/pgqueue/admin/redact.py b/src/pgqueue/admin/redact.py new file mode 100644 index 0000000..4f4ac00 --- /dev/null +++ b/src/pgqueue/admin/redact.py @@ -0,0 +1,35 @@ +"""What the public, read-only dashboard may show about errors and workers. + +The database keeps the full traceback and the real worker id (``host:pid:random``); +an operator needs both. A public viewer needs neither, and both leak the machine: +file system layout, interpreter version, host name and process ids. +""" + +from __future__ import annotations + +import hashlib +import re + +# An absolute POSIX path (two or more segments) or a Windows drive path, at a word start. +_PATH = re.compile(r"(? str: + """A short label that is stable for a worker id but reveals nothing about the host.""" + return "w-" + hashlib.sha256(worker_id.encode()).hexdigest()[:6] + + +def public_error(error: str | None) -> str | None: + """The last line of ``error`` (exception type and message) without paths or worker ids.""" + if not error: + return None + lines = error.strip().splitlines() + line = lines[-1].strip() if lines else "" + line = _WORKER_ID.sub(lambda m: worker_label(m.group()), line) + line = _PATH.sub("", line) + if len(line) > MAX_PUBLIC_ERROR: + line = line[: MAX_PUBLIC_ERROR - 3] + "..." + return line or None diff --git a/src/pgqueue/admin/schemas.py b/src/pgqueue/admin/schemas.py index 9405e93..66f404a 100644 --- a/src/pgqueue/admin/schemas.py +++ b/src/pgqueue/admin/schemas.py @@ -51,8 +51,9 @@ class ThroughputPoint(BaseModel): class WorkerOut(BaseModel): id: str - hostname: str - pid: int + hostname: str | None + """``None`` on the public dashboard, which shows a stable label as ``id`` instead.""" + pid: int | None queues: list[str] concurrency: int lease_seconds: float diff --git a/src/pgqueue/admin/static/dashboard.js b/src/pgqueue/admin/static/dashboard.js index 8b5d728..88d7193 100644 --- a/src/pgqueue/admin/static/dashboard.js +++ b/src/pgqueue/admin/static/dashboard.js @@ -236,7 +236,7 @@ const age = (now - new Date(w.last_seen_at)) / 1000; return el("article", { class: `worker ${w.alive ? "" : "lost"}` }, el("div", { class: "top" }, - el("span", { class: "wid", title: `${w.id} · ${w.hostname} pid ${w.pid}` }, shortId(w.id)), + el("span", { class: "wid", title: w.hostname ? `${w.id} · ${w.hostname} pid ${w.pid}` : w.id }, shortId(w.id)), el("span", { class: "state" }, w.alive ? "alive" : "lost")), el("div", { class: "meta-line" }, el("span", {}, `heartbeat ${age < 1.5 ? "just now" : `${age.toFixed(0)}s ago`}`), diff --git a/tests/test_admin_redaction.py b/tests/test_admin_redaction.py new file mode 100644 index 0000000..b6b863a --- /dev/null +++ b/tests/test_admin_redaction.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator + +import pytest +import pytest_asyncio +from fastapi import FastAPI + +from pgqueue import JobStatus, PgQueue +from pgqueue.admin import create_app +from pgqueue.admin.redact import public_error, worker_label + +from .asgi import Client + +TRACEBACK = ( + "Traceback (most recent call last):\n" + ' File "/data/data/com.termux/files/home/pgqueue/demo/tasks.py", line 12, in charge\n' + " raise RuntimeError('boom')\n" + "RuntimeError: card declined for /data/data/com.termux/files/usr/lib/python3.13/x.py" +) +WORKER = "localhost:4242:abc123" + + +def test_public_error_keeps_only_type_and_message_without_paths() -> None: + error = public_error(TRACEBACK) + assert error == "RuntimeError: card declined for " + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + (None, None), + ("", None), + (" \n ", None), + ("ValueError: ratio 1/2 is odd", "ValueError: ratio 1/2 is odd"), + ("OSError: C:\\Users\\me\\x.txt missing", "OSError: missing"), + ( + "lease expired: worker host.name:99:0a1b2c", + f"lease expired: worker {worker_label('host.name:99:0a1b2c')}", + ), + ], +) +def test_public_error_edge_cases(raw: str | None, expected: str | None) -> None: + assert public_error(raw) == expected + + +def test_public_error_is_bounded() -> None: + assert len(public_error("E: " + "x" * 5000) or "") <= 300 + + +def test_worker_label_is_stable_and_hides_the_host() -> None: + assert worker_label(WORKER) == worker_label(WORKER) + assert worker_label(WORKER) != worker_label("localhost:4242:abc124") + assert "localhost" not in worker_label(WORKER) + assert "4242" not in worker_label(WORKER) + + +@pytest_asyncio.fixture +async def public(app: PgQueue) -> AsyncIterator[FastAPI]: + admin_app = create_app(app.dsn, read_only=True) + yield admin_app + await admin_app.state.engine.dispose() + + +@pytest_asyncio.fixture +async def private(app: PgQueue) -> AsyncIterator[FastAPI]: + admin_app = create_app(app.dsn, read_only=False) + yield admin_app + await admin_app.state.engine.dispose() + + +async def _dead_job_on_worker(app: PgQueue) -> int: + job_id = await app.enqueue("demo.task", {"n": 1}, max_attempts=1) + [job] = await app.store.claim(["default"], limit=1, worker_id=WORKER, lease_seconds=30) + await app.store.register_worker(WORKER, queues=["default"], concurrency=1, lease_seconds=30) + await app.store.fail(job, WORKER, status=JobStatus.DEAD, error=TRACEBACK) + return job_id + + +async def test_public_jobs_api_shows_only_the_last_error_line( + app: PgQueue, public: FastAPI +) -> None: + job_id = await _dead_job_on_worker(app) + client = Client(public) + + for path in ("/api/jobs", f"/api/jobs/{job_id}"): + body = (await client.get(path)).text + assert "termux" not in body + assert "Traceback" not in body + assert "RuntimeError: card declined for " in body + assert "localhost" not in body + assert (await client.get(f"/api/jobs/{job_id}")).json()["locked_by"] == worker_label(WORKER) + + +async def test_public_live_overview_and_metrics_hide_host_details( + app: PgQueue, public: FastAPI +) -> None: + await _dead_job_on_worker(app) + client = Client(public) + + for path in ("/api/live", "/api/overview", "/metrics"): + body = (await client.get(path)).text + assert "termux" not in body, path + assert "localhost" not in body, path + assert "4242" not in body, path + assert worker_label(WORKER) in body or path != "/api/overview" + [worker] = (await client.get("/api/live")).json()["workers"] + assert worker["id"] == worker_label(WORKER) + assert worker["hostname"] is None + assert worker["pid"] is None + + +async def test_the_full_detail_stays_in_the_database_and_the_admin_api( + app: PgQueue, private: FastAPI +) -> None: + job_id = await _dead_job_on_worker(app) + client = Client(private) + + job = (await client.get(f"/api/jobs/{job_id}")).json() + assert job["last_error"] == TRACEBACK + assert job["locked_by"] == WORKER + [worker] = (await client.get("/api/live")).json()["workers"] + assert worker["id"] == WORKER + assert worker["hostname"] + assert isinstance(worker["pid"], int) From 54fc7a91a3f39029bbc40cf6242c7295b742c025 Mon Sep 17 00:00:00 2001 From: MaciejZiel Date: Wed, 14 Oct 2026 01:58:14 +0200 Subject: [PATCH 2/7] Reject ids outside the bigint range with 422 instead of a 500; add read-only DSN to create_app --- src/pgqueue/admin/api.py | 47 +++++++++++++++++++++++++++++++++------- tests/test_admin_api.py | 19 ++++++++++++++++ 2 files changed, 58 insertions(+), 8 deletions(-) diff --git a/src/pgqueue/admin/api.py b/src/pgqueue/admin/api.py index 14529d5..7b0ba46 100644 --- a/src/pgqueue/admin/api.py +++ b/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) @@ -209,6 +231,7 @@ def create_app( read_only: bool | None = None, banner: str | None = None, public: bool | None = None, + read_only_dsn: str | None = None, ) -> FastAPI: """Build the admin application. @@ -218,6 +241,11 @@ 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. ``public`` (default: ``PGQUEUE_ADMIN_PUBLIC``) is for an instance shown to anyone on @@ -231,6 +259,8 @@ def create_app( read_only = os.environ.get("PGQUEUE_ADMIN_READ_ONLY", "").lower() in ("1", "true", "yes") if public is None: public = os.environ.get("PGQUEUE_ADMIN_PUBLIC", "").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) @@ -250,6 +280,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: ) 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: @@ -283,7 +314,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) diff --git a/tests/test_admin_api.py b/tests/test_admin_api.py index 5143608..5a6b19d 100644 --- a/tests/test_admin_api.py +++ b/tests/test_admin_api.py @@ -2,6 +2,7 @@ from collections.abc import AsyncIterator +import pytest import pytest_asyncio from fastapi import FastAPI from pydantic import BaseModel @@ -174,3 +175,21 @@ async def test_icons_are_served_with_image_types(client: Client) -> None: assert icon.headers["content-type"].split(";")[0] == ctype assert icon.body assert (await client.get("/favicon.ico")).headers["cache-control"] == "public, max-age=86400" + + +@pytest.mark.parametrize("value", ["99999999999999999999", "0", "-1", str(2**63)]) +async def test_ids_outside_the_bigint_range_are_rejected_not_a_server_error( + client: Client, value: str +) -> None: + assert (await client.get("/api/jobs", {"before_id": value})).status == 422 + assert (await client.get(f"/api/jobs/{value}")).status == 422 + assert ( + await client.request("POST", f"/api/jobs/{value}/requeue", headers=ACTION) + ).status == 422 + + +async def test_the_largest_valid_id_is_simply_not_found(client: Client) -> None: + assert (await client.get(f"/api/jobs/{2**63 - 1}")).status == 404 + page = await client.get("/api/jobs", {"before_id": 2**63 - 1}) + assert page.status == 200 + assert page.json()["jobs"] == [] From a2c5ad18371898b8906d3781bdbc7736c6e78473 Mon Sep 17 00:00:00 2001 From: MaciejZiel Date: Wed, 14 Oct 2026 01:58:14 +0200 Subject: [PATCH 3/7] Let the demo's public dashboard connect with a separate read-only DSN --- demo/config.py | 3 +++ demo/server.py | 8 +++++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/demo/config.py b/demo/config.py index be8e8fc..fd78515 100644 --- a/demo/config.py +++ b/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, diff --git a/demo/server.py b/demo/server.py index e632c41..0dabb22 100644 --- a/demo/server.py +++ b/demo/server.py @@ -25,7 +25,13 @@ 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, public=True, banner=settings.banner) + app = create_app( + settings.dsn, + read_only=True, + public=True, + banner=settings.banner, + read_only_dsn=settings.read_only_dsn, + ) inner = app.router.lifespan_context @asynccontextmanager From 24f3ea3cae2eef2d27ad134a9c2bdfdb5c42b079 Mon Sep 17 00:00:00 2001 From: MaciejZiel Date: Wed, 14 Oct 2026 01:58:24 +0200 Subject: [PATCH 4/7] Cap the demo log with size-based rotation instead of appending forever --- deploy/termux/boot/pgqueue-demo | 7 +++-- deploy/termux/logcap.sh | 33 +++++++++++++++++++ tests/test_deploy_scripts.py | 56 +++++++++++++++++++++++++++++++++ 3 files changed, 94 insertions(+), 2 deletions(-) create mode 100755 deploy/termux/logcap.sh create mode 100644 tests/test_deploy_scripts.py diff --git a/deploy/termux/boot/pgqueue-demo b/deploy/termux/boot/pgqueue-demo index de351d2..3826b30 100755 --- a/deploy/termux/boot/pgqueue-demo +++ b/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 diff --git a/deploy/termux/logcap.sh b/deploy/termux/logcap.sh new file mode 100755 index 0000000..81dd4ab --- /dev/null +++ b/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 diff --git a/tests/test_deploy_scripts.py b/tests/test_deploy_scripts.py new file mode 100644 index 0000000..92aedb0 --- /dev/null +++ b/tests/test_deploy_scripts.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import shutil +import subprocess +from pathlib import Path + +import pytest + +DEPLOY = Path(__file__).resolve().parent.parent / "deploy" / "termux" + +pytestmark = pytest.mark.skipif(shutil.which("bash") is None, reason="needs bash") + + +def _logcap(log: Path, text: str, max_bytes: int, keep: int) -> None: + subprocess.run( + ["bash", str(DEPLOY / "logcap.sh"), str(log), str(max_bytes), str(keep)], + input=text.encode(), + check=True, + timeout=60, + ) + + +def test_logcap_keeps_the_log_bounded(tmp_path: Path) -> None: + log = tmp_path / "demo.log" + line = "x" * 99 + "\n" + _logcap(log, line * 5000, max_bytes=10_000, keep=2) + + files = sorted(p.name for p in tmp_path.iterdir()) + assert files == ["demo.log", "demo.log.1", "demo.log.2"] + # Rotation is checked every 50 lines, so a file may overshoot by at most that much. + for name in files: + assert (tmp_path / name).stat().st_size <= 10_000 + 50 * len(line) + # The newest output is in the live file. + assert log.read_text().endswith(line) + + +def test_logcap_without_backups_truncates(tmp_path: Path) -> None: + log = tmp_path / "demo.log" + _logcap(log, ("y" * 99 + "\n") * 2000, max_bytes=5_000, keep=0) + + assert [p.name for p in tmp_path.iterdir()] == ["demo.log"] + assert log.stat().st_size <= 5_000 + 50 * 100 + + +def test_logcap_keeps_a_last_line_without_newline(tmp_path: Path) -> None: + log = tmp_path / "demo.log" + _logcap(log, "first\nlast", max_bytes=1_000_000, keep=2) + assert log.read_text() == "first\nlast\n" + + +def test_boot_script_uses_logcap_and_all_scripts_parse() -> None: + boot = (DEPLOY / "boot" / "pgqueue-demo").read_text() + assert "logcap.sh" in boot + assert ">>" not in boot + for script in (DEPLOY / "boot" / "pgqueue-demo", DEPLOY / "logcap.sh", DEPLOY / "setup-db.sh"): + subprocess.run(["bash", "-n", str(script)], check=True) From b7b21846ec9c9af1378bc7d718a28f69c8328c0a Mon Sep 17 00:00:00 2001 From: MaciejZiel Date: Wed, 14 Oct 2026 01:59:34 +0200 Subject: [PATCH 5/7] Add a SELECT-only database role, connection limits and REVOKE CONNECT to setup-db.sh --- deploy/termux/env.example | 7 +++++++ deploy/termux/setup-db.sh | 38 +++++++++++++++++++++++++++++++++----- 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/deploy/termux/env.example b/deploy/termux/env.example index ab09a78..831aafe 100644 --- a/deploy/termux/env.example +++ b/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:change-me@127.0.0.1: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:change-me-too@127.0.0.1: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 diff --git a/deploy/termux/setup-db.sh b/deploy/termux/setup-db.sh index 1a10d1b..7161bc3 100755 --- a/deploy/termux/setup-db.sh +++ b/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)" From 637a2be6443302c39523fbd51060304859ad048d Mon Sep 17 00:00:00 2001 From: MaciejZiel Date: Wed, 14 Oct 2026 01:59:34 +0200 Subject: [PATCH 6/7] Test the read-only DSN and the database setup script --- tests/test_readonly_role.py | 146 ++++++++++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) create mode 100644 tests/test_readonly_role.py diff --git a/tests/test_readonly_role.py b/tests/test_readonly_role.py new file mode 100644 index 0000000..3d8207a --- /dev/null +++ b/tests/test_readonly_role.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import asyncio +import os +import shutil +import subprocess +from collections.abc import AsyncIterator +from pathlib import Path +from urllib.parse import urlsplit, urlunsplit + +import asyncpg +import pytest +import pytest_asyncio + +from pgqueue import PgQueue +from pgqueue.admin import create_app + +from .asgi import Client +from .conftest import dsn_for_database + +ROLE = "pgqueue_test_ro" +SETUP = Path(__file__).resolve().parent.parent / "deploy" / "termux" / "setup-db.sh" + + +def _as_role(dsn: str, user: str, password: str, database: str | None = None) -> str: + parts = urlsplit(dsn) + host = f"{parts.hostname}:{parts.port}" if parts.port else str(parts.hostname) + path = f"/{database}" if database else parts.path + return urlunsplit(parts._replace(netloc=f"{user}:{password}@{host}", path=path)) + + +@pytest_asyncio.fixture +async def readonly_dsn(app: PgQueue) -> AsyncIterator[str]: + """A login role that can only SELECT from the test database.""" + database = urlsplit(app.dsn).path.lstrip("/") + conn = await asyncpg.connect(app.dsn) + try: + await conn.execute(f"DROP ROLE IF EXISTS {ROLE}") + await conn.execute(f"CREATE ROLE {ROLE} LOGIN PASSWORD 'ro'") + await conn.execute(f'GRANT CONNECT ON DATABASE "{database}" TO {ROLE}') + await conn.execute(f"GRANT USAGE ON SCHEMA public TO {ROLE}") + await conn.execute(f"GRANT SELECT ON ALL TABLES IN SCHEMA public TO {ROLE}") + yield _as_role(app.dsn, ROLE, "ro") + finally: + await conn.execute(f"DROP OWNED BY {ROLE}") + await conn.execute(f"DROP ROLE IF EXISTS {ROLE}") + await conn.close() + + +async def test_a_read_only_app_serves_the_dashboard_through_the_select_only_role( + app: PgQueue, readonly_dsn: str +) -> None: + await app.enqueue("demo.task", {"n": 1}) + admin = create_app( + "postgresql://nobody:wrong@127.0.0.1:1/none", read_only=True, read_only_dsn=readonly_dsn + ) + try: + client = Client(admin) + assert (await client.get("/api/live")).status == 200 + assert (await client.get("/api/jobs")).json()["jobs"][0]["task"] == "demo.task" + assert (await client.get("/metrics")).status == 200 + finally: + await admin.state.engine.dispose() + + conn = await asyncpg.connect(readonly_dsn) + try: + with pytest.raises(asyncpg.InsufficientPrivilegeError): + await conn.execute("UPDATE pgqueue_jobs SET status = 'failed'") + with pytest.raises(asyncpg.InsufficientPrivilegeError): + await conn.execute("DELETE FROM pgqueue_jobs") + finally: + await conn.close() + + +async def test_the_read_only_dsn_only_applies_to_a_read_only_app( + app: PgQueue, readonly_dsn: str +) -> None: + admin = create_app(app.dsn, read_only=False, read_only_dsn="postgresql://x:y@127.0.0.1:1/none") + try: + assert (await Client(admin).get("/api/live")).status == 200 + finally: + await admin.state.engine.dispose() + + +async def test_the_read_only_dsn_can_come_from_the_environment( + app: PgQueue, readonly_dsn: str, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("PGQUEUE_READONLY_DSN", readonly_dsn) + admin = create_app("postgresql://nobody:wrong@127.0.0.1:1/none", read_only=True) + try: + assert (await Client(admin).get("/api/live")).status == 200 + finally: + await admin.state.engine.dispose() + + +@pytest.mark.skipif(shutil.which("psql") is None, reason="needs the psql client") +async def test_setup_script_creates_a_limited_owner_and_a_select_only_role( + database: str, +) -> None: + parts = urlsplit(database) + env = { + **os.environ, + "PGHOST": str(parts.hostname), + "PGPORT": str(parts.port or 5432), + "PGUSER": str(parts.username), + "PGPASSWORD": str(parts.password), + "DEMO_DB_PASSWORD": "owner-pw", + "DEMO_RO_PASSWORD": "ro-pw", + } + admin = await asyncpg.connect(dsn_for_database(database, "postgres")) + try: + for _ in range(2): # idempotent + await asyncio.to_thread( + subprocess.run, ["bash", str(SETUP)], env=env, check=True, capture_output=True + ) + limits = dict( + await admin.fetch( + "SELECT rolname, rolconnlimit FROM pg_roles WHERE rolname LIKE 'pgqueue_demo%'" + ) + ) + assert limits == {"pgqueue_demo": 40, "pgqueue_demo_ro": 10} + + owner_dsn = _as_role(database, "pgqueue_demo", "owner-pw", "pgqueue_demo") + ro_dsn = _as_role(database, "pgqueue_demo_ro", "ro-pw", "pgqueue_demo") + owner = await asyncpg.connect(owner_dsn) + await owner.execute("CREATE TABLE later (id int)") # created after the grants + await owner.execute("INSERT INTO later VALUES (1)") + reader = await asyncpg.connect(ro_dsn) + try: + assert await reader.fetchval("SELECT count(*) FROM later") == 1 + with pytest.raises(asyncpg.PostgresError): + await reader.execute("INSERT INTO later VALUES (2)") + with pytest.raises(asyncpg.PostgresError): + await reader.execute("CREATE TABLE x (id int)") + finally: + await reader.close() + await owner.close() + # No other role can connect to the demo database by default. + assert not await admin.fetchval( + "SELECT has_database_privilege('public', 'pgqueue_demo', 'CONNECT')" + ) + finally: + await admin.execute("DROP DATABASE IF EXISTS pgqueue_demo WITH (FORCE)") + await admin.execute("DROP ROLE IF EXISTS pgqueue_demo_ro") + await admin.execute("DROP ROLE IF EXISTS pgqueue_demo") + await admin.close() From 60a88c0c44164487879f3713acb055db944d9e0e Mon Sep 17 00:00:00 2001 From: MaciejZiel Date: Wed, 14 Oct 2026 01:59:34 +0200 Subject: [PATCH 7/7] Document the public API redaction, read-only role and log rotation --- CHANGELOG.md | 11 +++++++++++ README.md | 24 +++++++++++++++++++++--- 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 43bf276..54823d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,8 +38,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 diff --git a/README.md b/README.md index 8371382..cb81132 100644 --- a/README.md +++ b/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 ``, 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`. + **What a visitor can and cannot do.** The demo has no input path for code or URLs. Every job is built by the demo's own generator from fixed task types and numbers, and the "webhook" tasks only sleep and fail on a schedule: nothing makes an outgoing request. The public app @@ -281,7 +291,9 @@ $ 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 @@ -301,8 +313,14 @@ $ deploy/termux/start.sh # try it by hand fi Nothing else changes: `nginx-zones.conf` and `~/.config/pgqueue-demo/env` stay as they are. - **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