diff --git a/CHANGELOG.md b/CHANGELOG.md index ab6cab6..a7d58d5 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 950fb0d..87a98fb 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`. + ```console $ export PGQUEUE_DSN=postgresql://user:pass@127.0.0.1/pgqueue_demo $ python -m demo # migrates, then serves http://127.0.0.1:8080 @@ -265,7 +275,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 @@ -273,8 +285,14 @@ $ deploy/termux/start.sh # try it by hand fi `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 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 904c3cf..f95f019 100644 --- a/demo/server.py +++ b/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 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/env.example b/deploy/termux/env.example index d33f6ac..c32502e 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/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/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)" diff --git a/src/pgqueue/admin/api.py b/src/pgqueue/admin/api.py index 36aa7db..e2bb687 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) @@ -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) 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 6bb2c5d..fa945d1 100644 --- a/src/pgqueue/admin/static/dashboard.js +++ b/src/pgqueue/admin/static/dashboard.js @@ -235,7 +235,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_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"] == [] 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) 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) 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()