From 7f1eec9bf6dd72b0551853fc2db84ff3785e6461 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Fri, 10 Jul 2026 15:41:44 +0200 Subject: [PATCH 1/6] feat(dashboard): unstable health badge + /metrics exposure warning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ported from improvements made in CashPilot-Desktop (the Go port), applied where the web app didn't already have them: - dashboard: surface crashes_7d + an "unstable" flag (>=3 crashes in the 7-day health window) on each deployed service row, and render a distinct "unstable" badge (worst tone + explicit label + restarts/crashes in the tooltip) so a crash-looping earner is legible at a glance, not just via a lower score number. Reuses the existing get_health_scores data (already computes crashes/restarts). - metrics: log a startup WARNING when Prometheus /metrics is enabled — it is served unauthenticated and exposes earnings/health, so keep it behind a proxy and off untrusted networks. Tests: two new deployed-row tests (unstable at/above vs below the threshold). ruff clean; full suite 997 passed. --- app/main.py | 6 +++++ app/metrics.py | 12 +++++++++ app/static/js/app.js | 12 +++++++-- tests/test_main_routes.py | 56 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 84 insertions(+), 2 deletions(-) diff --git a/app/main.py b/app/main.py index b1c361f..756089e 100644 --- a/app/main.py +++ b/app/main.py @@ -528,6 +528,10 @@ async def api_services_deployed(request: Request) -> list[dict[str, Any]]: "health_score": health.get("score"), "uptime_pct": health.get("uptime_pct"), "restarts_7d": health.get("restarts", 0), + "crashes_7d": health.get("crashes", 0), + # "unstable" flags a service that has crashed repeatedly in the health window + # so the dashboard can surface it at a glance (not just via the score number). + "unstable": health.get("crashes", 0) >= 3, "instances": len(agg["instances"]), "instance_details": instance_details, "collector_disconnected": slug in alert_slugs, @@ -567,6 +571,8 @@ async def api_services_deployed(request: Request) -> list[dict[str, Any]]: "health_score": None, "uptime_pct": None, "restarts_7d": 0, + "crashes_7d": 0, + "unstable": False, "instances": 0, "instance_details": [], "collector_disconnected": slug in alert_slugs, diff --git a/app/metrics.py b/app/metrics.py index 9046bc1..cbe52d4 100644 --- a/app/metrics.py +++ b/app/metrics.py @@ -17,12 +17,15 @@ from __future__ import annotations import contextlib +import logging import os import re import time from fastapi import FastAPI +logger = logging.getLogger(__name__) + METRICS_ENABLED = os.getenv("CASHPILOT_METRICS_ENABLED", "").lower() in ("1", "true", "yes") _registry = None @@ -231,6 +234,15 @@ def setup(app: FastAPI) -> None: _init_metrics() + # /metrics is served UNAUTHENTICATED (Prometheus convention). It exposes earnings and + # health totals, so warn on startup: keep it behind a reverse proxy / auth and never + # expose the app's port directly to an untrusted network. + logger.warning( + "Prometheus /metrics is ENABLED and served UNAUTHENTICATED — it exposes your " + "earnings and health data to anyone who can reach this port. Keep it behind a " + "reverse proxy with auth; do not expose the port directly to an untrusted network." + ) + from fastapi import Response from prometheus_client import CONTENT_TYPE_LATEST, generate_latest from starlette.middleware.base import BaseHTTPMiddleware diff --git a/app/static/js/app.js b/app/static/js/app.js index 4458645..3085f43 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -434,8 +434,16 @@ const CP = (() => { let healthBadge = '--'; if (!isExternal && svc.health_score !== null && svc.health_score !== undefined) { const score = svc.health_score; - const hClass = score >= 80 ? 'badge-running' : score >= 50 ? 'badge-error' : 'badge-stopped'; - healthBadge = `${score}`; + const crashes = svc.crashes_7d || 0; + const restarts = svc.restarts_7d || 0; + const unstable = svc.unstable === true; + // Unstable (repeated crashes in the 7-day window) takes the worst tone + an explicit + // label so a crash-looping service is legible at a glance, not just via a low number. + const hClass = (unstable || score < 50) ? 'badge-stopped' : score >= 80 ? 'badge-running' : 'badge-error'; + const uptime = (svc.uptime_pct !== null && svc.uptime_pct !== undefined) ? `${svc.uptime_pct}% uptime · ` : ''; + const title = `Health ${score}/100 · ${uptime}${restarts} restarts · ${crashes} crashes (7d)`; + const label = unstable ? `unstable · ${crashes} crashes` : String(score); + healthBadge = `${label}`; } // Balance + delta from breakdown diff --git a/tests/test_main_routes.py b/tests/test_main_routes.py index 8b0ff6e..6dba698 100644 --- a/tests/test_main_routes.py +++ b/tests/test_main_routes.py @@ -639,6 +639,62 @@ def test_deployed_collector_not_needs_setup_when_creds_present(self, client): row = next(r for r in resp.json() if r["slug"] == "repocket") assert row["collector_needs_setup"] is False + def test_deployed_row_flags_unstable_on_repeated_crashes(self, client): + """A service with >=3 crashes in the health window is flagged unstable so the + dashboard can surface a crash-looping earner at a glance.""" + workers = [ + { + "id": 1, + "name": "w1", + "status": "online", + "system_info": json.dumps({"docker_available": True}), + "containers": json.dumps([{"slug": "repocket", "name": "rp", "status": "running"}]), + "apps": "[]", + } + ] + scores = [{"slug": "repocket", "score": 30, "uptime_pct": 40, "restarts": 5, "crashes": 4}] + with ( + _auth_owner(), + patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers), + patch("app.main.database.get_earnings_summary", new_callable=AsyncMock, return_value=[]), + patch("app.main.database.get_health_scores", new_callable=AsyncMock, return_value=scores), + patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=[]), + patch("app.main.database.get_config", new_callable=AsyncMock, return_value={}), + patch("app.main.catalog.get_service", return_value={"name": "Repocket", "category": "bandwidth"}), + ): + resp = client.get("/api/services/deployed") + assert resp.status_code == 200 + row = next(r for r in resp.json() if r["slug"] == "repocket") + assert row["unstable"] is True + assert row["crashes_7d"] == 4 + + def test_deployed_row_not_unstable_below_threshold(self, client): + """Fewer than 3 crashes is not flagged unstable (crashes_7d is still surfaced).""" + workers = [ + { + "id": 1, + "name": "w1", + "status": "online", + "system_info": json.dumps({"docker_available": True}), + "containers": json.dumps([{"slug": "repocket", "name": "rp", "status": "running"}]), + "apps": "[]", + } + ] + scores = [{"slug": "repocket", "score": 85, "uptime_pct": 98, "restarts": 1, "crashes": 2}] + with ( + _auth_owner(), + patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers), + patch("app.main.database.get_earnings_summary", new_callable=AsyncMock, return_value=[]), + patch("app.main.database.get_health_scores", new_callable=AsyncMock, return_value=scores), + patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=[]), + patch("app.main.database.get_config", new_callable=AsyncMock, return_value={}), + patch("app.main.catalog.get_service", return_value={"name": "Repocket", "category": "bandwidth"}), + ): + resp = client.get("/api/services/deployed") + row = next(r for r in resp.json() if r["slug"] == "repocket") + assert row["unstable"] is False + assert row["crashes_7d"] == 2 + def test_deployed_disconnected_takes_precedence_over_needs_setup(self, client): # When the collector has actually errored, show the error state, not "needs setup". workers = [ From 25a79735b85c7644b9fc11a0f3cbc85179c46b13 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Fri, 10 Jul 2026 16:18:29 +0200 Subject: [PATCH 2/6] =?UTF-8?q?fix(security):=20close=20code-audit=20findi?= =?UTF-8?q?ngs=20=E2=80=94=20container=20escape,=20event-loop=20stalls,=20?= =?UTF-8?q?authz,=20error=20surfacing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From a full-codebase 8-lens audit. Security + correctness fixes: Security: - worker_api: replace the bypassable deploy-spec volume DENYLIST with a realpath-normalized check that blocks /var/run + /run parents (closing the /var/run:/x -> docker.sock -> host-root escape) and adds a network_mode allowlist (permits host for mysterium, blocks container:). Authenticate the previously-unauthenticated worker mini-UI status page. - main: deploy via /api/workers/{id}/command is now owner-gated (was writer — an owner-gate bypass of /api/deploy). Single-source the secret-config-key set from database.SECRET_CONFIG_KEYS (the shorter local list left remember_web/xsrf_token unmasked in /api/collectors/meta). - auth/routers: invalidate outstanding sessions on user role-change and deletion (a demoted/deleted user kept access until the token expired) via the existing per-user epoch mechanism. - app.js: sanitizeHint rejects non-http(s)/mailto hrefs (blocks javascript:). Correctness (event-loop stalls under async): - main: _validate_worker_url's DNS resolution now runs via asyncio.to_thread at its sole call site (a slow-DNS worker no longer stalls the whole UI event loop); the validator stays sync so its existing tests are unaffected. - worker_api: 4 unthreaded docker_available() calls now use asyncio.to_thread (they blocked exactly when Docker was degraded). - main: fire-and-forget collection tasks keep a reference + log exceptions (_spawn); record_collection_start moved inside the try; the hourly collector gather is bounded by a semaphore (anti-storm as the catalog grows). Consistency / error-surfacing: - main: both container-action route families now record BOTH the health event and the metrics lifecycle (they had diverged). - exchange_rates: wire the previously-dead rates_stale() into get_all() so the UI can flag stale FX. - app.js: stop fabricating $0.00 / eternal spinners / a false-healthy bell on fetch errors — surface the error, keep last values; remove dead code. Tests: negative-role test for deploy-via-command (owner-only); updated the route side-effect + deploy-gate tests to the new behavior; added a _login_attempts autouse reset fixture (removes a latent order-dependent flake). Verified: ruff clean; full pytest 998 passed. --- app/database.py | 8 ++- app/exchange_rates.py | 1 + app/main.py | 88 +++++++++++++++++++++++--------- app/routers/users.py | 9 ++++ app/static/js/app.js | 76 +++++++++++++++++---------- app/worker_api.py | 46 +++++++++++++---- tests/conftest.py | 18 +++++++ tests/test_coverage_gaps.py | 2 +- tests/test_main_deploy_routes.py | 26 +++++++++- 9 files changed, 209 insertions(+), 65 deletions(-) diff --git a/app/database.py b/app/database.py index e94c0bd..a73ef06 100644 --- a/app/database.py +++ b/app/database.py @@ -89,7 +89,13 @@ def decrypt_value(value: str) -> str: try: return _fernet.decrypt(value[len(_ENC_PREFIX) :].encode()).decode() except InvalidToken: - _logger.warning("Failed to decrypt config value — key may have changed") + _logger.warning( + "Failed to decrypt config value: the Fernet ENCRYPTION KEY (CASHPILOT_SECRET_KEY / " + "%s) does not match the key this value was encrypted with. This is NOT a bad " + "credential — re-enter the affected credentials, or restore the original encryption " + "key, to recover.", + _FERNET_KEY_FILE, + ) return "" diff --git a/app/exchange_rates.py b/app/exchange_rates.py index 8e879a6..765db3b 100644 --- a/app/exchange_rates.py +++ b/app/exchange_rates.py @@ -108,6 +108,7 @@ def get_all() -> dict[str, Any]: "fiat": dict(_fiat_rates), "crypto_usd": dict(_crypto_usd), "last_updated": _last_fetch, + "stale": rates_stale(), } diff --git a/app/main.py b/app/main.py index 756089e..14b3479 100644 --- a/app/main.py +++ b/app/main.py @@ -44,6 +44,29 @@ # In-memory store for the latest collector alerts (errors from last run) _collector_alerts: list[dict[str, str]] = [] _collection_lock = asyncio.Lock() +_collection_semaphore = asyncio.Semaphore(8) + +# Fire-and-forget background tasks (e.g. triggered collection runs). Keeping a +# reference prevents the task from being garbage-collected mid-run and lets us +# retrieve/log any exception it raised (bare `asyncio.create_task(...)` drops +# the reference and silently swallows exceptions). +_background_tasks: set[asyncio.Task] = set() + + +def _spawn(coro) -> asyncio.Task: + """Fire-and-forget a coroutine while keeping a reference and logging errors.""" + task = asyncio.create_task(coro) + _background_tasks.add(task) + + def _on_done(t: asyncio.Task) -> None: + _background_tasks.discard(t) + if not t.cancelled(): + exc = t.exception() + if exc is not None: + logger.error("Background task failed: %s", exc, exc_info=exc) + + task.add_done_callback(_on_done) + return task # Login rate limiting _login_attempts: dict[str, list[float]] = defaultdict(list) @@ -179,6 +202,12 @@ async def _run_health_check() -> None: logger.warning("Health check skipped: %s", exc) +async def _collect_bounded(collector) -> Any: + """Run a single collector's `collect()` under the shared concurrency limit.""" + async with _collection_semaphore: + return await collector.collect() + + async def _run_collection() -> None: """Collect earnings from all deployed services that have collectors.""" global _collector_alerts @@ -186,9 +215,10 @@ async def _run_collection() -> None: logger.info("Collection already in progress, skipping") return async with _collection_lock: - start_time = metrics.record_collection_start() success = True + start_time = 0.0 try: + start_time = metrics.record_collection_start() deployments = await database.get_deployments() config = await database.get_config() or {} if not isinstance(config, dict): @@ -197,7 +227,7 @@ async def _run_collection() -> None: collectors = make_collectors(deployments, config) await _close_stale() - results = await asyncio.gather(*(c.collect() for c in collectors), return_exceptions=True) + results = await asyncio.gather(*(_collect_bounded(c) for c in collectors), return_exceptions=True) alerts: list[dict[str, str]] = [] platforms_ok = 0 for result in results: @@ -335,7 +365,7 @@ def _on_job_event(event): ) scheduler.start() await exchange_rates.refresh() - asyncio.create_task(_run_collection()) + _spawn(_run_collection()) logger.info("CashPilot UI started (container ops via workers)") yield @@ -756,7 +786,7 @@ async def api_deploy(request: Request, slug: str, body: DeployRequest, worker_id await database.save_deployment(slug=slug, container_id=container_id) await database.record_health_event(slug, "start", f"deployed to worker {worker_id}") metrics.record_container_lifecycle("deploy", slug) - asyncio.create_task(_run_collection()) + _spawn(_run_collection()) return {"status": "deployed", "container_id": container_id} @@ -765,6 +795,7 @@ async def api_stop(request: Request, slug: str, worker_id: int | None = None) -> _require_writer(request) worker_id = await _resolve_worker_id(worker_id) result = await _proxy_worker_command(worker_id, "stop", slug) + await database.record_health_event(slug, "stop") metrics.record_container_lifecycle("stop", slug) return result @@ -774,6 +805,7 @@ async def api_restart(request: Request, slug: str, worker_id: int | None = None) _require_writer(request) worker_id = await _resolve_worker_id(worker_id) result = await _proxy_worker_command(worker_id, "restart", slug) + await database.record_health_event(slug, "restart") metrics.record_container_lifecycle("restart", slug) return result @@ -787,6 +819,7 @@ async def api_remove( params = {"delete_volumes": "true"} if delete_volumes else None result = await _proxy_worker_command(worker_id, "remove", slug, params=params) await database.remove_deployment(slug) + await database.record_health_event(slug, "remove") metrics.record_container_lifecycle("remove", slug) return result @@ -878,6 +911,9 @@ def _validate_worker_url(raw_url: str) -> str: Resolves hostnames and validates the resolved IP(s) so a DNS name that points at a metadata/loopback address is rejected (DNS-rebinding guard). + Synchronous — its only blocking op is DNS resolution; event-loop callers + MUST invoke it via ``asyncio.to_thread`` (see _get_verified_worker_url) so a + slow/hanging resolver never blocks the whole UI. """ parsed = urlparse(raw_url) if parsed.scheme not in _ALLOWED_WORKER_SCHEMES: @@ -924,7 +960,7 @@ def _validate_worker_url(raw_url: str) -> str: return raw_url.rstrip("/") -def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[str, str]]: +async def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[str, str]]: """Validate a worker record and return (url, headers).""" if not worker: raise HTTPException(status_code=404, detail="Worker not found") @@ -932,7 +968,7 @@ def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[str, str raise HTTPException(status_code=503, detail="Worker is offline") if not worker["url"]: raise HTTPException(status_code=503, detail="Worker URL not known") - url = _validate_worker_url(worker["url"]) + url = await asyncio.to_thread(_validate_worker_url, worker["url"]) headers: dict[str, str] = {} if FLEET_API_KEY: headers["Authorization"] = f"Bearer {FLEET_API_KEY}" @@ -944,7 +980,7 @@ async def _proxy_worker_command( ) -> dict[str, Any]: """Forward a container command (restart/stop/start/remove) to a worker.""" worker = await database.get_worker(worker_id) - url, headers = _get_verified_worker_url(worker) + url, headers = await _get_verified_worker_url(worker) try: async with httpx.AsyncClient(timeout=30) as client: @@ -964,7 +1000,7 @@ async def _proxy_worker_command( async def _proxy_worker_deploy(worker_id: int, slug: str, spec: dict[str, Any]) -> dict[str, Any]: """Forward a deploy command with full spec to a worker.""" worker = await database.get_worker(worker_id) - url, headers = _get_verified_worker_url(worker) + url, headers = await _get_verified_worker_url(worker) try: async with httpx.AsyncClient(timeout=60) as client: @@ -981,7 +1017,7 @@ async def _proxy_worker_deploy(worker_id: int, slug: str, spec: dict[str, Any]) async def _proxy_worker_logs(worker_id: int, slug: str, lines: int = 50) -> dict[str, str]: """Forward a logs request to a worker.""" worker = await database.get_worker(worker_id) - url, headers = _get_verified_worker_url(worker) + url, headers = await _get_verified_worker_url(worker) try: async with httpx.AsyncClient(timeout=30) as client: @@ -1010,6 +1046,7 @@ async def api_service_restart(request: Request, slug: str, worker_id: int | None worker_id = await _resolve_worker_id(worker_id) result = await _proxy_worker_command(worker_id, "restart", slug) await database.record_health_event(slug, "restart") + metrics.record_container_lifecycle("restart", slug) return result @@ -1019,6 +1056,7 @@ async def api_service_stop(request: Request, slug: str, worker_id: int | None = worker_id = await _resolve_worker_id(worker_id) result = await _proxy_worker_command(worker_id, "stop", slug) await database.record_health_event(slug, "stop") + metrics.record_container_lifecycle("stop", slug) return result @@ -1028,6 +1066,7 @@ async def api_service_start(request: Request, slug: str, worker_id: int | None = worker_id = await _resolve_worker_id(worker_id) result = await _proxy_worker_command(worker_id, "start", slug) await database.record_health_event(slug, "start") + metrics.record_container_lifecycle("start", slug) return result @@ -1049,6 +1088,8 @@ async def api_service_remove( params = {"delete_volumes": "true"} if delete_volumes else None result = await _proxy_worker_command(worker_id, "remove", slug, params=params) await database.remove_deployment(slug) + await database.record_health_event(slug, "remove") + metrics.record_container_lifecycle("remove", slug) return result @@ -1241,7 +1282,7 @@ async def api_health_scores(request: Request, days: int = 7) -> list[dict[str, A @app.post("/api/collect") async def api_collect(request: Request) -> dict[str, str]: _require_writer(request) - asyncio.create_task(_run_collection()) + _spawn(_run_collection()) return {"status": "collection_started"} @@ -1331,7 +1372,7 @@ async def api_set_preferences(request: Request, body: PreferencesUpdate) -> dict ) # If setup is completed, trigger an immediate collection if body.setup_completed: - asyncio.create_task(_run_collection()) + _spawn(_run_collection()) return {"status": "saved"} @@ -1409,17 +1450,11 @@ async def api_collectors_meta(request: Request) -> list[dict[str, Any]]: _require_owner(request) from app.collectors import _COLLECTOR_ARGS, COLLECTOR_MAP - secret_args = { - "password", - "token", - "auth_token", - "access_token", - "api_key", - "session_cookie", - "auth_cookie", - "oauth_token", - "brd_sess_id", - } + # Single-sourced from database.SECRET_CONFIG_KEYS so this endpoint can never + # disagree with the encryption-at-rest / masking logic about which config + # keys are secret (a hand-maintained duplicate here previously missed + # `remember_web` and `xsrf_token`, unmasking them). + secret_args = database.SECRET_CONFIG_KEYS # Per-service hints on how to obtain the credentials hints: dict[str, str] = { "bitping": "Use your Bitping account email and password (same as nodes.bitping.com).", @@ -1711,10 +1746,15 @@ class WorkerCommand(BaseModel): @app.post("/api/workers/{worker_id}/command") async def api_worker_command(request: Request, worker_id: int, body: WorkerCommand) -> dict[str, Any]: """Send a command to a worker by proxying to its REST API.""" - _require_writer(request) + # Deploy is owner-gated everywhere else (see /api/deploy/{slug}); a writer + # must not be able to bypass that gate by sending command="deploy" here. + if body.command == "deploy": + _require_owner(request) + else: + _require_writer(request) worker = await database.get_worker(worker_id) - url, headers = _get_verified_worker_url(worker) + url, headers = await _get_verified_worker_url(worker) try: async with httpx.AsyncClient(timeout=30) as client: diff --git a/app/routers/users.py b/app/routers/users.py index 2b2bed9..26f21f2 100644 --- a/app/routers/users.py +++ b/app/routers/users.py @@ -7,6 +7,7 @@ from __future__ import annotations +import time from typing import Any from fastapi import APIRouter, HTTPException, Request @@ -43,6 +44,10 @@ async def api_update_user_role(request: Request, user_id: int, body: UserRoleUpd if owner_count <= 1: raise HTTPException(status_code=400, detail="Cannot remove the last owner") await main.database.update_user_role(user_id, body.role) + # Invalidate any outstanding session tokens for this user — they carry the + # old role and must not be trusted until re-issued (same mechanism used + # for password changes). + main.auth.set_user_pwd_epoch(user_id, time.time()) return {"status": "updated"} @@ -55,4 +60,8 @@ async def api_delete_user(request: Request, user_id: int) -> dict[str, str]: if not user: raise HTTPException(status_code=404, detail="User not found") await main.database.delete_user(user_id) + # Invalidate any outstanding session tokens for the now-deleted account. + # The epoch cache is in-memory keyed by uid, so it survives the row being + # gone from the DB — decode_session_token rejects the token on iat alone. + main.auth.set_user_pwd_epoch(user_id, time.time()) return {"status": "deleted"} diff --git a/app/static/js/app.js b/app/static/js/app.js index 3085f43..fdfe957 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -75,13 +75,19 @@ const CP = (() => { function sanitizeHint(html) { const el = document.createElement('div'); el.innerHTML = html; + const allowedSchemes = ['http:', 'https:', 'mailto:']; el.querySelectorAll('*').forEach(node => { if (!['A', 'B', 'CODE'].includes(node.tagName)) { node.replaceWith(document.createTextNode(node.textContent)); return; } for (const attr of [...node.attributes]) { - if (node.tagName === 'A' && (attr.name === 'href' || attr.name === 'target')) continue; + if (node.tagName === 'A' && attr.name === 'href') { + const scheme = (attr.value || '').trim().toLowerCase(); + if (!allowedSchemes.some(s => scheme.startsWith(s))) node.removeAttribute('href'); + continue; + } + if (node.tagName === 'A' && attr.name === 'target') continue; node.removeAttribute(attr.name); } }); @@ -178,7 +184,9 @@ const CP = (() => { async function loadExchangeRates() { try { _exchangeRates = await api('/api/exchange-rates'); - } catch { /* keep defaults */ } + } catch (err) { + console.warn('Could not refresh exchange rates, keeping previous values:', err); + } } async function loadTopbarEarnings() { @@ -186,8 +194,9 @@ const CP = (() => { await loadExchangeRates(); const data = await api('/api/earnings/summary'); setTextContent('topbar-total', formatCurrency(data.total || 0)); - } catch { - setTextContent('topbar-total', formatCurrency(0)); + } catch (err) { + // Keep whatever was already shown rather than fabricating $0. + console.warn('Could not refresh topbar earnings, keeping previous value:', err); } } @@ -239,12 +248,9 @@ const CP = (() => { setChangeIndicator('month-change', data.month_change); } } catch (err) { - // API not yet implemented — fill with placeholder - setTextContent('total-earnings', formatCurrency(0)); - setTextContent('today-earnings', formatCurrency(0)); - setTextContent('month-earnings', formatCurrency(0)); - setTextContent('active-services', '0'); - setTextContent('topbar-total', formatCurrency(0)); + // Keep whatever was already displayed — a transient fetch failure + // must not look like the dashboard's earnings dropped to zero. + toast(err.message || 'Could not refresh dashboard', 'error'); } } @@ -401,9 +407,15 @@ const CP = (() => { } }); } catch (err) { - // Keep existing table if we had one; only show spinner on truly empty state + // Keep the existing table on a refresh failure. Only replace the + // perpetual "Loading..." spinner with a clear error + retry affordance + // when there was nothing on screen yet. if (!container.querySelector('.breakdown-table')) { - container.innerHTML = `
Loading services...
`; + container.innerHTML = ` +
+ Couldn't load services${err && err.message ? `: ${escapeHtml(err.message)}` : ''}. + +
`; } } } @@ -509,7 +521,7 @@ const CP = (() => { ? (eligible ? 'Cash out earnings' : 'View payout details') : 'No payout info available'; const claimDisabled = !co.dashboard_url; - const claimBtn = ``; @@ -528,7 +540,7 @@ const CP = (() => { // For single instance: show action buttons directly let actionBtns; if (isMulti) { - const chevron = ``; actionBtns = `
${claimBtn}${settingsBtn}${chevron}
`; @@ -544,13 +556,13 @@ const CP = (() => { ${claimBtn} ${settingsBtn} ${_canWrite ? ` - - - ` : ''} `; @@ -598,13 +610,13 @@ const CP = (() => {
${_canWrite ? ` - - - ` : ''}
@@ -826,7 +838,7 @@ const CP = (() => { inputs.forEach(input => { const key = input.dataset.config; const val = input.value.trim(); - if (val && val !== '********') data[key] = val; + if (val) data[key] = val; }); if (!Object.keys(data).length) { toast('Enter at least one credential', 'warning'); @@ -926,8 +938,6 @@ const CP = (() => { // ----------------------------------------------------------- // Claim Modal // ----------------------------------------------------------- - let _breakdownCache = []; - async function openClaimModal(platform) { openModal('claim-modal'); const title = document.getElementById('claim-modal-title'); @@ -1766,7 +1776,7 @@ const CP = (() => {
The credentials above run the service. To also see its balance on the dashboard, add earnings-tracking credentials under - Settings → Collectors + Settings → Collectors after deploying. This is optional — the service earns either way.
`; } @@ -1982,7 +1992,7 @@ const CP = (() => { const data = {}; inputs.forEach(input => { const val = input.value.trim(); - if (val && val !== '********') { + if (val) { data[input.dataset.envKey] = val; } }); @@ -2064,7 +2074,7 @@ const CP = (() => { inputs.forEach(input => { const key = input.dataset.config; const val = input.value.trim(); - if (val && val !== '********') { + if (val) { data[key] = val; } }); @@ -2202,6 +2212,10 @@ const CP = (() => { try { const alerts = await api('/api/collector-alerts'); + // Clear any muted "alerts unavailable" styling left over from a prior + // failed poll now that the fetch succeeded. + badge.style.background = ''; + badge.title = ''; if (!alerts || alerts.length === 0) { badge.style.display = 'none'; list.innerHTML = '
All collectors healthy
'; @@ -2223,7 +2237,14 @@ const CP = (() => { `).join(''); } catch { - badge.style.display = 'none'; + // The bell's own fetch failed — this is unknown, not healthy. Show a + // neutral/muted indicator rather than hiding the badge (which would + // read as "all collectors healthy"). + badge.style.display = ''; + badge.style.background = 'var(--text-muted)'; + badge.textContent = '?'; + badge.title = 'Alerts unavailable'; + list.innerHTML = '
Alerts unavailable — could not reach the server
'; } } @@ -2433,6 +2454,7 @@ const CP = (() => { filterCatalog, refreshServices, openClaimModal, + loadServicesTable, toggleInstances, deployServiceToWorkers, workerAction, diff --git a/app/worker_api.py b/app/worker_api.py index 5777057..5979c6d 100644 --- a/app/worker_api.py +++ b/app/worker_api.py @@ -94,7 +94,7 @@ async def _send_heartbeat() -> None: "os": f"{platform.system()} {platform.release()}", "arch": platform.machine(), "hostname": socket.gethostname(), - "docker_available": orchestrator.docker_available(), + "docker_available": await asyncio.to_thread(orchestrator.docker_available), }, } @@ -148,7 +148,7 @@ async def lifespan(app: FastAPI): global _heartbeat_task logger.info("CashPilot Worker '%s' starting", WORKER_NAME) - docker_mode = "direct" if orchestrator.docker_available() else "monitor-only" + docker_mode = "direct" if await asyncio.to_thread(orchestrator.docker_available) else "monitor-only" logger.info("Docker: %s", docker_mode) if UI_URL: @@ -177,8 +177,9 @@ async def lifespan(app: FastAPI): @app.get("/", response_class=HTMLResponse) -async def worker_status_page(): +async def worker_status_page(request: Request): """Self-contained HTML status page for the worker.""" + _verify_api_key(request) containers = [] try: containers = await asyncio.to_thread(orchestrator.get_status_cached) @@ -237,7 +238,7 @@ async def worker_status_page():
{_esc(WORKER_NAME)}
{_esc(socket.gethostname())}
{_esc(platform.system())} {_esc(platform.machine())}
-
{"Available" if orchestrator.docker_available() else "Not available"}
+
{"Available" if await asyncio.to_thread(orchestrator.docker_available) else "Not available"}
{ui_status}
{_last_heartbeat}
@@ -285,8 +286,29 @@ class DeploySpec(BaseModel): resources: ResourceSpec | None = None -_BLOCKED_VOLUME_SOURCES = {"/", "/etc", "/var/run/docker.sock", "/root", "/proc", "/sys"} +_BLOCKED_VOLUME_ROOTS = { + "/", + "/etc", + "/root", + "/proc", + "/sys", + "/boot", + "/dev", + "/var", + "/usr", + "/home", + "/lib", + "/lib64", + "/bin", + "/sbin", + "/var/run", + "/run", + "/var/lib/docker", +} _BLOCKED_CAPS = {"ALL", "SYS_ADMIN", "SYS_PTRACE"} +# Named volumes (bridge/none/unset) and mysterium's legitimate host networking are allowed; +# `container:` (namespace join) and any other value are rejected. +_ALLOWED_NETWORK_MODES = {None, "", "bridge", "none", "host"} # Docker memory size syntax: a positive integer with an optional b/k/m/g unit. _MEM_LIMIT_RE = re.compile(r"^\d+[bkmgBKMG]?$") @@ -298,12 +320,14 @@ def _validate_deploy_spec(spec: DeploySpec) -> None: blocked = _BLOCKED_CAPS & {c.upper() for c in spec.cap_add} if blocked: raise HTTPException(status_code=403, detail=f"Blocked capabilities: {', '.join(blocked)}") + if spec.network_mode not in _ALLOWED_NETWORK_MODES: + raise HTTPException(status_code=403, detail=f"Network mode '{spec.network_mode}' is not allowed") for source in spec.volumes: - normalized = "/" + source.strip("/") - if normalized == "/": - raise HTTPException(status_code=403, detail=f"Volume mount '{source}' is blocked") - for blocked in _BLOCKED_VOLUME_SOURCES: - if normalized == blocked or normalized.startswith(blocked + "/"): + if not source.startswith("/"): + continue # named volume (e.g. "mysterium-data") — always allowed + real = os.path.realpath(source) + for blocked in _BLOCKED_VOLUME_ROOTS: + if real == blocked or real.startswith(blocked + "/"): raise HTTPException(status_code=403, detail=f"Volume mount '{source}' is blocked") _validate_resources(spec.resources) @@ -335,7 +359,7 @@ async def api_worker_status(request: Request) -> dict[str, Any]: logger.warning("Failed to get container status: %s", exc) return { "name": WORKER_NAME, - "docker_available": orchestrator.docker_available(), + "docker_available": await asyncio.to_thread(orchestrator.docker_available), "ui_connected": _ui_connected, "container_count": len(containers), "running_count": sum(1 for c in containers if c.get("status") == "running"), diff --git a/tests/conftest.py b/tests/conftest.py index 3418065..1c86c0f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -44,3 +44,21 @@ async def _drain(): # No usable loop (e.g. one is already running) — best-effort cleanup. with contextlib.suppress(RuntimeError): asyncio.run(_drain()) + + +@pytest.fixture(autouse=True) +def _reset_login_attempts(): + """Clear the in-process login rate-limit bucket before every test. + + ``app.main._login_attempts`` is a module-level dict keyed by client host that + persists for the whole test process, and TestClient's host is a constant + ("testclient"), so failed-login attempts from one test would otherwise leak + into later tests that hit /login — an order-dependent landmine (and the reason + a real rate-limit test couldn't be added safely before). Start each test with + an empty bucket. + """ + with contextlib.suppress(Exception): + from app import main + + main._login_attempts.clear() + yield diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py index 3725f1c..19f2faf 100644 --- a/tests/test_coverage_gaps.py +++ b/tests/test_coverage_gaps.py @@ -705,7 +705,7 @@ def test_worker_command_deploy_error(self, client): mock_client.post.return_value = error_resp with ( - _auth_writer(), + _auth_owner(), # deploy via command is owner-gated patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker), patch("app.main.httpx.AsyncClient", return_value=mock_client), patch("app.main.FLEET_API_KEY", "test-key"), diff --git a/tests/test_main_deploy_routes.py b/tests/test_main_deploy_routes.py index e0493ed..0938d36 100644 --- a/tests/test_main_deploy_routes.py +++ b/tests/test_main_deploy_routes.py @@ -343,6 +343,7 @@ def test_remove_service(self, client): patch("app.main.database.remove_deployment", new_callable=AsyncMock), patch("app.main.httpx.AsyncClient", return_value=mock_client), patch("app.main.FLEET_API_KEY", "test-key"), + patch("app.main.database.record_health_event", new_callable=AsyncMock), ): resp = client.delete("/api/services/honeygain") assert resp.status_code == 200 @@ -374,6 +375,7 @@ def test_old_stop_route(self, client): patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker), patch("app.main.httpx.AsyncClient", return_value=mock_client), patch("app.main.FLEET_API_KEY", "test-key"), + patch("app.main.database.record_health_event", new_callable=AsyncMock), ): resp = client.post("/api/stop/honeygain") assert resp.status_code == 200 @@ -386,6 +388,7 @@ def test_old_restart_route(self, client): patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker), patch("app.main.httpx.AsyncClient", return_value=mock_client), patch("app.main.FLEET_API_KEY", "test-key"), + patch("app.main.database.record_health_event", new_callable=AsyncMock), ): resp = client.post("/api/restart/honeygain") assert resp.status_code == 200 @@ -399,6 +402,7 @@ def test_old_remove_route(self, client): patch("app.main.database.remove_deployment", new_callable=AsyncMock), patch("app.main.httpx.AsyncClient", return_value=mock_client), patch("app.main.FLEET_API_KEY", "test-key"), + patch("app.main.database.record_health_event", new_callable=AsyncMock), ): resp = client.delete("/api/remove/honeygain") assert resp.status_code == 200 @@ -412,6 +416,7 @@ def test_remove_with_delete_volumes(self, client): patch("app.main.database.remove_deployment", new_callable=AsyncMock), patch("app.main.httpx.AsyncClient", return_value=mock_client), patch("app.main.FLEET_API_KEY", "test-key"), + patch("app.main.database.record_health_event", new_callable=AsyncMock), ): resp = client.delete("/api/services/honeygain?delete_volumes=true") assert resp.status_code == 200 @@ -507,9 +512,11 @@ def _setup(self): return worker, mock_client def test_command_deploy(self, client): + # Deploy via the command route is OWNER-gated (matching /api/deploy/{slug}), + # so it must not be reachable with a mere writer role — use owner here. worker, mock_client = self._setup() with ( - _auth_writer(), + _auth_owner(), patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker), patch("app.main.httpx.AsyncClient", return_value=mock_client), patch("app.main.FLEET_API_KEY", "test-key"), @@ -524,6 +531,23 @@ def test_command_deploy(self, client): ) assert resp.status_code == 200 + def test_command_deploy_writer_denied(self, client): + """A writer must NOT be able to deploy via /api/workers/{id}/command — deploy is + owner-only, so this route must not become an owner-gate bypass (writer stays + allowed for stop/restart/remove, tested separately).""" + worker, mock_client = self._setup() + with ( + _auth_writer(), + patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker), + patch("app.main.httpx.AsyncClient", return_value=mock_client), + patch("app.main.FLEET_API_KEY", "test-key"), + ): + resp = client.post( + "/api/workers/1/command", + json={"command": "deploy", "slug": "honeygain", "spec": {"image": "test"}}, + ) + assert resp.status_code == 403 + def test_command_stop(self, client): worker, mock_client = self._setup() with ( From d316f4ae2f096608059b05c6fc5ca0e4e9631d9e Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:42:26 +0200 Subject: [PATCH 3/6] style: apply ruff format to main.py (CI runs ruff format --check) --- app/main.py | 1 + 1 file changed, 1 insertion(+) diff --git a/app/main.py b/app/main.py index 14b3479..e948b3a 100644 --- a/app/main.py +++ b/app/main.py @@ -68,6 +68,7 @@ def _on_done(t: asyncio.Task) -> None: task.add_done_callback(_on_done) return task + # Login rate limiting _login_attempts: dict[str, list[float]] = defaultdict(list) _LOGIN_MAX_ATTEMPTS = 5 From efd274ea7ddb2cad3ebe5e1afaae7f765204d554 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:17:34 +0200 Subject: [PATCH 4/6] =?UTF-8?q?perf(db):=20bound=20health=5Fevents=20growt?= =?UTF-8?q?h=20=E2=80=94=20split=20retention=20+=20weekly=20VACUUM?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit health_events is dominated by check_ok/check_down uptime samples (one per service every 5 minutes), yet get_health_scores only reads a bounded window (/api/health/scores caps it at 90 days) while retention kept everything 400 days. Trim samples to HEALTH_CHECK_RETENTION_DAYS=95 (just past the max query window, so no allowed query can out-range its own samples) while keeping the rare lifecycle events (start/stop/restart/crash) the full 400 days. Add a weekly VACUUM job (commit-first, WAL-checkpoint TRUNCATE) since SQLite never shrinks the file on DELETE alone — without it the DB keeps its high-water-mark size forever. Uptime% is unchanged for every allowed query. Tests cover the split-retention boundary + a clean VACUUM run. From c9ac3882e065fbfb04d708b3dd2d8b43753c8d68 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:24:25 +0200 Subject: [PATCH 5/6] test: cover orchestrator/worker_api Docker paths; expect db_vacuum job Add 61 tests (Docker SDK fully mocked) for the two lowest-covered, most privileged modules: orchestrator.py 38%->73%, worker_api.py 44%->74%. Covers _collect_stats (zero-delta CPU, missing keys, APIError), _find_container name/ label/ValueError fallback, get_status[_light] docker-unavailable + corrupted- container-skip paths, _verify_api_key negative paths, _validate_deploy_spec rejections (privileged, cap_add, network_mode, blocked volume roots incl. subpaths; host/bridge + named volumes allowed), and every command endpoint's success + ValueError->404 / RuntimeError->503 branches. Also update the lifespan-scheduler test to expect the new db_vacuum interval job (added in the previous commit). --- tests/test_main_routes.py | 3 +- tests/test_orchestrator_coverage.py | 301 +++++++++++++++++++++++++ tests/test_worker_api_coverage.py | 333 ++++++++++++++++++++++++++++ 3 files changed, 636 insertions(+), 1 deletion(-) create mode 100644 tests/test_orchestrator_coverage.py create mode 100644 tests/test_worker_api_coverage.py diff --git a/tests/test_main_routes.py b/tests/test_main_routes.py index 6dba698..c66de19 100644 --- a/tests/test_main_routes.py +++ b/tests/test_main_routes.py @@ -1800,12 +1800,13 @@ async def _run(): async with main_mod.lifespan(main_mod.app): sched = main_mod.scheduler jobs = {j.id: j for j in sched.get_jobs()} - # All five interval jobs registered. + # All interval jobs registered. assert set(jobs) == { "collect", "health_check", "stale_workers", "data_retention", + "db_vacuum", "exchange_rates", } # Every job carries the hardening kwargs (audit fix). diff --git a/tests/test_orchestrator_coverage.py b/tests/test_orchestrator_coverage.py new file mode 100644 index 0000000..8a9f23b --- /dev/null +++ b/tests/test_orchestrator_coverage.py @@ -0,0 +1,301 @@ +"""Tests targeting uncovered lines in app/orchestrator.py. + +Covers get_status / get_status_light (running / exited / missing-container / +docker-unavailable branches and the image-matched external-container path), +_collect_stats CPU/memory parsing (including zero-delta and error edge +cases), and _find_container's label-based fallback lookup. + +Mocks the Docker SDK entirely — no real Docker socket is used. +""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock, patch + +os.environ.setdefault("CASHPILOT_API_KEY", "test-fleet-key") + +import pytest # noqa: E402 + +docker = pytest.importorskip("docker") # noqa: E402 + +from docker.errors import APIError, NotFound # noqa: E402 + +from app import orchestrator # noqa: E402 +from app.constants import LABEL_MANAGED, LABEL_SERVICE # noqa: E402 + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _mock_container(*, name, status, slug, deployed_by="worker", category="bandwidth", container_id="short123"): + c = MagicMock() + c.id = f"id-{name}" + c.name = name + c.status = status + c.short_id = container_id + c.labels = { + LABEL_SERVICE: slug, + LABEL_MANAGED: "true", + "cashpilot.deployed-by": deployed_by, + "cashpilot.category": category, + } + c.image.tags = [f"{slug}:latest"] + c.image.short_id = "sha256:abcdef" + c.attrs = {"Created": "2026-01-01T00:00:00Z"} + return c + + +def _zero_stats(): + return { + "cpu_stats": {"cpu_usage": {"total_usage": 1, "percpu_usage": [1]}, "system_cpu_usage": 10}, + "precpu_stats": {"cpu_usage": {"total_usage": 0}, "system_cpu_usage": 5}, + "memory_stats": {"usage": 0}, + } + + +# --------------------------------------------------------------------------- +# _collect_stats +# --------------------------------------------------------------------------- + + +class TestCollectStats: + def test_parses_cpu_and_memory(self): + c = MagicMock() + c.stats.return_value = { + "cpu_stats": { + "cpu_usage": {"total_usage": 2_000_000_000, "percpu_usage": [1, 1]}, + "system_cpu_usage": 100_000_000_000, + "online_cpus": 2, + }, + "precpu_stats": { + "cpu_usage": {"total_usage": 1_000_000_000}, + "system_cpu_usage": 90_000_000_000, + }, + "memory_stats": {"usage": 209_715_200}, # 200 MB + } + cpu_pct, mem_mb = orchestrator._collect_stats(c) + assert cpu_pct == 20.0 + assert mem_mb == 200.0 + + def test_zero_system_delta_returns_zero_cpu(self): + c = MagicMock() + c.stats.return_value = { + "cpu_stats": {"cpu_usage": {"total_usage": 500, "percpu_usage": [1]}, "system_cpu_usage": 500}, + "precpu_stats": {"cpu_usage": {"total_usage": 500}, "system_cpu_usage": 500}, + "memory_stats": {"usage": 0}, + } + cpu_pct, mem_mb = orchestrator._collect_stats(c) + assert cpu_pct == 0.0 + assert mem_mb == 0.0 + + def test_missing_memory_usage_defaults_to_zero(self): + c = MagicMock() + c.stats.return_value = { + "cpu_stats": {"cpu_usage": {"total_usage": 100, "percpu_usage": [1]}, "system_cpu_usage": 500}, + "precpu_stats": {"cpu_usage": {"total_usage": 50}, "system_cpu_usage": 400}, + "memory_stats": {}, # no "usage" key + } + _, mem_mb = orchestrator._collect_stats(c) + assert mem_mb == 0.0 + + def test_missing_key_returns_zero_tuple(self): + c = MagicMock() + c.stats.return_value = {"cpu_stats": {}} # precpu_stats missing -> KeyError + assert orchestrator._collect_stats(c) == (0.0, 0.0) + + def test_stats_api_error_returns_zero_tuple(self): + c = MagicMock() + c.stats.side_effect = APIError("stats unavailable") + assert orchestrator._collect_stats(c) == (0.0, 0.0) + + +# --------------------------------------------------------------------------- +# _find_container +# --------------------------------------------------------------------------- + + +class TestFindContainer: + def test_finds_by_name(self): + container = MagicMock() + client = MagicMock() + client.containers.get.return_value = container + with patch.object(orchestrator, "_get_client", return_value=client): + result = orchestrator._find_container("honeygain") + assert result is container + client.containers.get.assert_called_once_with("cashpilot-honeygain") + client.containers.list.assert_not_called() + + def test_falls_back_to_label_lookup_when_renamed(self): + container = MagicMock() + client = MagicMock() + client.containers.get.side_effect = NotFound("nope") + client.containers.list.return_value = [container] + with patch.object(orchestrator, "_get_client", return_value=client): + result = orchestrator._find_container("honeygain") + assert result is container + _, kwargs = client.containers.list.call_args + assert kwargs["filters"]["label"] == [ + f"{LABEL_SERVICE}=honeygain", + f"{LABEL_MANAGED}=true", + ] + + def test_raises_value_error_when_not_found_anywhere(self): + client = MagicMock() + client.containers.get.side_effect = NotFound("nope") + client.containers.list.return_value = [] + with ( + patch.object(orchestrator, "_get_client", return_value=client), + pytest.raises(ValueError, match="honeygain"), + ): + orchestrator._find_container("honeygain") + + +# --------------------------------------------------------------------------- +# get_status (slow path, includes CPU/mem stats) +# --------------------------------------------------------------------------- + + +class TestGetStatus: + def test_docker_unavailable_returns_empty(self): + with patch.object(orchestrator, "_get_client", side_effect=RuntimeError("no socket")): + assert orchestrator.get_status() == [] + + def test_running_container_reports_stats(self): + container = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain") + container.stats.return_value = { + "cpu_stats": {"cpu_usage": {"total_usage": 100, "percpu_usage": [1]}, "system_cpu_usage": 500}, + "precpu_stats": {"cpu_usage": {"total_usage": 50}, "system_cpu_usage": 400}, + "memory_stats": {"usage": 1_048_576}, + } + client = MagicMock() + client.containers.list.return_value = [container] + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(orchestrator, "_build_image_slug_map", return_value={}), + ): + results = orchestrator.get_status() + assert len(results) == 1 + row = results[0] + assert row["slug"] == "honeygain" + assert row["status"] == "running" + assert row["memory_mb"] == 1.0 + + def test_exited_container_missing_stats_handled_gracefully(self): + container = _mock_container(name="cashpilot-earnapp", status="exited", slug="earnapp") + container.stats.side_effect = APIError("exited, no stats") + client = MagicMock() + client.containers.list.return_value = [container] + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(orchestrator, "_build_image_slug_map", return_value={}), + ): + results = orchestrator.get_status() + assert results[0]["status"] == "exited" + assert results[0]["cpu_percent"] == 0.0 + assert results[0]["memory_mb"] == 0.0 + + def test_corrupted_container_is_skipped_not_crashed(self): + good = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain") + good.stats.return_value = _zero_stats() + bad = MagicMock() + bad.id = "bad-id" + bad.short_id = "bad12" + bad.labels.get.side_effect = RuntimeError("corrupted container labels") + client = MagicMock() + client.containers.list.return_value = [bad, good] + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(orchestrator, "_build_image_slug_map", return_value={}), + ): + results = orchestrator.get_status() + assert len(results) == 1 + assert results[0]["slug"] == "honeygain" + + def test_image_matched_external_container_included(self): + external = MagicMock() + external.id = "ext-1" + external.name = "manually-run-storj" + external.status = "running" + external.image.tags = ["storjlabs/storagenode:latest"] + external.short_id = "ext1" + external.attrs = {"Created": "2026-01-01T00:00:00Z"} + external.stats.return_value = _zero_stats() + client = MagicMock() + client.containers.list.side_effect = [[], [external]] + image_map = {"storjlabs/storagenode:latest": "storj"} + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(orchestrator, "_build_image_slug_map", return_value=image_map), + ): + results = orchestrator.get_status() + assert len(results) == 1 + assert results[0]["slug"] == "storj" + assert results[0]["deployed_by"] == "external" + + def test_all_containers_listing_failure_still_returns_labeled(self): + labeled = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain") + labeled.stats.return_value = _zero_stats() + client = MagicMock() + client.containers.list.side_effect = [[labeled], Exception("docker daemon hiccup")] + image_map = {"some/image:latest": "svc"} + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(orchestrator, "_build_image_slug_map", return_value=image_map), + ): + results = orchestrator.get_status() + assert len(results) == 1 + assert results[0]["slug"] == "honeygain" + + +# --------------------------------------------------------------------------- +# get_status_light (fast path, no CPU/mem stats) +# --------------------------------------------------------------------------- + + +class TestGetStatusLight: + def test_docker_unavailable_returns_empty(self): + with patch.object(orchestrator, "_get_client", side_effect=RuntimeError("no socket")): + assert orchestrator.get_status_light() == [] + + def test_no_stats_call_even_when_running(self): + container = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain") + client = MagicMock() + client.containers.list.return_value = [container] + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(orchestrator, "_build_image_slug_map", return_value={}), + ): + results = orchestrator.get_status_light() + assert results[0]["cpu_percent"] == 0.0 + assert results[0]["memory_mb"] == 0.0 + container.stats.assert_not_called() + + def test_image_matched_container_skipped_when_slug_already_seen(self): + labeled = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain") + dup = MagicMock() + dup.id = "dup-id" + dup.image.tags = ["honeygain/desktop:latest"] + client = MagicMock() + client.containers.list.side_effect = [[labeled], [dup]] + image_map = {"honeygain/desktop:latest": "honeygain"} + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(orchestrator, "_build_image_slug_map", return_value=image_map), + ): + results = orchestrator.get_status_light() + # The duplicate slug from the image-matched scan must not be added again. + assert len(results) == 1 + + def test_all_containers_listing_failure_handled(self): + labeled = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain") + client = MagicMock() + client.containers.list.side_effect = [[labeled], Exception("boom")] + image_map = {"x": "y"} + with ( + patch.object(orchestrator, "_get_client", return_value=client), + patch.object(orchestrator, "_build_image_slug_map", return_value=image_map), + ): + results = orchestrator.get_status_light() + assert len(results) == 1 diff --git a/tests/test_worker_api_coverage.py b/tests/test_worker_api_coverage.py new file mode 100644 index 0000000..7246e24 --- /dev/null +++ b/tests/test_worker_api_coverage.py @@ -0,0 +1,333 @@ +"""Tests targeting uncovered lines in app/worker_api.py. + +Covers _verify_api_key negative paths (missing header, wrong key, no key +configured), _validate_deploy_spec rejections not already exercised by +test_worker_resources.py (privileged, blocked capabilities, disallowed +network_mode, blocked volume roots, named-volume allow-list), and the +container command endpoints (status/list/deploy/restart/stop/start/remove/ +logs) success + docker-error paths. + +Mocks app.orchestrator entirely — no real Docker socket is used. +""" + +from __future__ import annotations + +import os +from contextlib import asynccontextmanager +from unittest.mock import MagicMock, patch + +os.environ.setdefault("CASHPILOT_API_KEY", "test-fleet-key") + +import pytest # noqa: E402 + +try: + from fastapi import HTTPException # noqa: E402 + from fastapi.testclient import TestClient # noqa: E402 + + from app import worker_api # noqa: E402 + from app.worker_api import DeploySpec, _validate_deploy_spec, _verify_api_key # noqa: E402 +except ImportError: + pytest.skip( + "Requires full app dependencies (fastapi, docker, etc.) — runs in CI", + allow_module_level=True, + ) + + +# --------------------------------------------------------------------------- +# Helpers — match the TestClient harness used in test_worker_resources.py +# --------------------------------------------------------------------------- + + +@asynccontextmanager +async def _noop_lifespan(a): + yield + + +def _client(): + # Disable the heartbeat lifespan so the TestClient stays isolated. + worker_api.app.router.lifespan_context = _noop_lifespan + return TestClient(worker_api.app, raise_server_exceptions=False) + + +def _auth(): + return {"Authorization": f"Bearer {worker_api.API_KEY}"} + + +# --------------------------------------------------------------------------- +# _verify_api_key — negative paths +# --------------------------------------------------------------------------- + + +class TestVerifyApiKeyUnit: + def test_missing_header_rejected(self): + req = MagicMock() + req.headers = {} + with pytest.raises(HTTPException) as ei: + _verify_api_key(req) + assert ei.value.status_code == 401 + + def test_wrong_key_rejected(self): + req = MagicMock() + req.headers = {"Authorization": "Bearer wrong-key"} + with pytest.raises(HTTPException) as ei: + _verify_api_key(req) + assert ei.value.status_code == 401 + + def test_no_key_configured_returns_503(self): + req = MagicMock() + req.headers = {"Authorization": "Bearer anything"} + with patch.object(worker_api, "API_KEY", ""), pytest.raises(HTTPException) as ei: + _verify_api_key(req) + assert ei.value.status_code == 503 + + def test_correct_key_passes(self): + req = MagicMock() + req.headers = {"Authorization": f"Bearer {worker_api.API_KEY}"} + _verify_api_key(req) # must not raise + + +class TestEndpointAuthRejection: + def test_endpoint_rejects_missing_header(self): + resp = _client().get("/api/status") + assert resp.status_code == 401 + + def test_endpoint_rejects_wrong_key(self): + resp = _client().get("/api/status", headers={"Authorization": "Bearer nope"}) + assert resp.status_code == 401 + + def test_health_endpoint_requires_no_auth(self): + resp = _client().get("/api/health") + assert resp.status_code == 200 + assert resp.json() == {"status": "ok", "worker": worker_api.WORKER_NAME} + + +# --------------------------------------------------------------------------- +# _validate_deploy_spec — rejections not covered by test_worker_resources.py +# --------------------------------------------------------------------------- + + +class TestValidateDeploySpecRejections: + def test_privileged_rejected(self): + spec = DeploySpec(image="x", privileged=True) + with pytest.raises(HTTPException) as ei: + _validate_deploy_spec(spec) + assert ei.value.status_code == 403 + + def test_blocked_capability_rejected(self): + spec = DeploySpec(image="x", cap_add=["SYS_ADMIN"]) + with pytest.raises(HTTPException) as ei: + _validate_deploy_spec(spec) + assert ei.value.status_code == 403 + + def test_blocked_capability_rejected_case_insensitive(self): + spec = DeploySpec(image="x", cap_add=["sys_ptrace"]) + with pytest.raises(HTTPException) as ei: + _validate_deploy_spec(spec) + assert ei.value.status_code == 403 + + def test_allowed_capability_passes(self): + _validate_deploy_spec(DeploySpec(image="x", cap_add=["NET_ADMIN"])) # must not raise + + def test_disallowed_network_mode_rejected(self): + spec = DeploySpec(image="x", network_mode="container:abc123") + with pytest.raises(HTTPException) as ei: + _validate_deploy_spec(spec) + assert ei.value.status_code == 403 + + def test_host_network_mode_allowed(self): + # mysterium legitimately needs host networking. + _validate_deploy_spec(DeploySpec(image="x", network_mode="host")) # must not raise + + def test_bridge_network_mode_allowed(self): + _validate_deploy_spec(DeploySpec(image="x", network_mode="bridge")) # must not raise + + @pytest.mark.parametrize("root", ["/etc", "/root", "/var/run", "/proc"]) + def test_blocked_volume_root_rejected(self, root): + # Patch realpath to identity: on macOS dev machines /etc and /var/run + # are themselves symlinks (-> /private/etc, /private/var/run), which + # would dodge the block by resolving to a path outside the blocklist. + # In the worker's actual Linux container this resolution is a no-op. + spec = DeploySpec(image="x", volumes={root: {"bind": "/data", "mode": "rw"}}) + with ( + patch("app.worker_api.os.path.realpath", side_effect=lambda p: p), + pytest.raises(HTTPException) as ei, + ): + _validate_deploy_spec(spec) + assert ei.value.status_code == 403 + + def test_blocked_volume_subpath_rejected(self): + spec = DeploySpec(image="x", volumes={"/etc/passwd": {"bind": "/x", "mode": "ro"}}) + with ( + patch("app.worker_api.os.path.realpath", side_effect=lambda p: p), + pytest.raises(HTTPException) as ei, + ): + _validate_deploy_spec(spec) + assert ei.value.status_code == 403 + + def test_named_volume_always_allowed(self): + # Not an absolute path -> named volume (e.g. mysterium-data), always allowed. + spec = DeploySpec(image="x", volumes={"mysterium-data": {"bind": "/data", "mode": "rw"}}) + _validate_deploy_spec(spec) # must not raise + + def test_allowed_bind_mount_passes(self): + spec = DeploySpec(image="x", volumes={"/data/honeygain": {"bind": "/data", "mode": "rw"}}) + _validate_deploy_spec(spec) # must not raise + + +# --------------------------------------------------------------------------- +# Container command endpoints — success + docker-error paths +# --------------------------------------------------------------------------- + + +class TestStatusAndListEndpoints: + def test_status_reports_docker_and_counts(self): + containers = [{"status": "running"}, {"status": "exited"}] + with ( + patch("app.worker_api.orchestrator.get_status_cached", return_value=containers), + patch("app.worker_api.orchestrator.docker_available", return_value=True), + ): + resp = _client().get("/api/status", headers=_auth()) + assert resp.status_code == 200 + body = resp.json() + assert body["docker_available"] is True + assert body["container_count"] == 2 + assert body["running_count"] == 1 + + def test_status_handles_status_fetch_exception(self): + with ( + patch("app.worker_api.orchestrator.get_status_cached", side_effect=Exception("boom")), + patch("app.worker_api.orchestrator.docker_available", return_value=False), + ): + resp = _client().get("/api/status", headers=_auth()) + assert resp.status_code == 200 + assert resp.json()["container_count"] == 0 + + def test_list_containers_success(self): + containers = [{"slug": "honeygain"}] + with patch("app.worker_api.orchestrator.get_status_cached", return_value=containers): + resp = _client().get("/api/containers", headers=_auth()) + assert resp.status_code == 200 + assert resp.json() == containers + + def test_list_containers_docker_unavailable_returns_503(self): + with patch("app.worker_api.orchestrator.get_status_cached", side_effect=RuntimeError("no docker")): + resp = _client().get("/api/containers", headers=_auth()) + assert resp.status_code == 503 + + +class TestDeployEndpointErrors: + def test_generic_exception_returns_500(self): + with patch("app.worker_api.orchestrator.deploy_raw", side_effect=Exception("boom")): + resp = _client().post( + "/api/containers/honeygain/deploy", + json={"image": "img"}, + headers=_auth(), + ) + assert resp.status_code == 500 + + +class TestStopRestartStartEndpoints: + def test_stop_success(self): + with patch("app.worker_api.orchestrator.stop_service") as mock_stop: + resp = _client().post("/api/containers/honeygain/stop", headers=_auth()) + assert resp.status_code == 200 + assert resp.json() == {"status": "stopped"} + mock_stop.assert_called_once_with("honeygain") + + def test_stop_not_found_returns_404(self): + with patch("app.worker_api.orchestrator.stop_service", side_effect=ValueError("not found")): + resp = _client().post("/api/containers/missing/stop", headers=_auth()) + assert resp.status_code == 404 + + def test_stop_docker_unavailable_returns_503(self): + with patch("app.worker_api.orchestrator.stop_service", side_effect=RuntimeError("no docker")): + resp = _client().post("/api/containers/honeygain/stop", headers=_auth()) + assert resp.status_code == 503 + + def test_restart_success(self): + with patch("app.worker_api.orchestrator.restart_service") as mock_restart: + resp = _client().post("/api/containers/honeygain/restart", headers=_auth()) + assert resp.status_code == 200 + assert resp.json() == {"status": "restarted"} + mock_restart.assert_called_once_with("honeygain") + + def test_restart_not_found_returns_404(self): + with patch("app.worker_api.orchestrator.restart_service", side_effect=ValueError("gone")): + resp = _client().post("/api/containers/x/restart", headers=_auth()) + assert resp.status_code == 404 + + def test_restart_docker_unavailable_returns_503(self): + with patch("app.worker_api.orchestrator.restart_service", side_effect=RuntimeError("no docker")): + resp = _client().post("/api/containers/honeygain/restart", headers=_auth()) + assert resp.status_code == 503 + + def test_start_success(self): + with patch("app.worker_api.orchestrator.start_service") as mock_start: + resp = _client().post("/api/containers/honeygain/start", headers=_auth()) + assert resp.status_code == 200 + assert resp.json() == {"status": "started"} + mock_start.assert_called_once_with("honeygain") + + def test_start_not_found_returns_404(self): + with patch("app.worker_api.orchestrator.start_service", side_effect=ValueError("gone")): + resp = _client().post("/api/containers/x/start", headers=_auth()) + assert resp.status_code == 404 + + def test_start_docker_unavailable_returns_503(self): + with patch("app.worker_api.orchestrator.start_service", side_effect=RuntimeError("no docker")): + resp = _client().post("/api/containers/honeygain/start", headers=_auth()) + assert resp.status_code == 503 + + +class TestRemoveEndpoint: + def test_remove_success(self): + removal = {"container": "cashpilot-honeygain", "deleted_volumes": [], "failed_volumes": []} + with patch("app.worker_api.orchestrator.remove_service", return_value=removal) as mock_remove: + resp = _client().delete("/api/containers/honeygain", headers=_auth()) + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "removed" + assert body["container"] == "cashpilot-honeygain" + mock_remove.assert_called_once_with("honeygain", delete_volumes=False) + + def test_remove_with_delete_volumes_query_param(self): + removal = {"container": "cashpilot-honeygain", "deleted_volumes": ["v1"], "failed_volumes": []} + with patch("app.worker_api.orchestrator.remove_service", return_value=removal) as mock_remove: + resp = _client().delete("/api/containers/honeygain?delete_volumes=true", headers=_auth()) + assert resp.status_code == 200 + mock_remove.assert_called_once_with("honeygain", delete_volumes=True) + + def test_remove_not_found_returns_404(self): + with patch("app.worker_api.orchestrator.remove_service", side_effect=ValueError("gone")): + resp = _client().delete("/api/containers/x", headers=_auth()) + assert resp.status_code == 404 + + def test_remove_docker_unavailable_returns_503(self): + with patch("app.worker_api.orchestrator.remove_service", side_effect=RuntimeError("no docker")): + resp = _client().delete("/api/containers/honeygain", headers=_auth()) + assert resp.status_code == 503 + + +class TestLogsEndpoint: + def test_logs_success(self): + with patch("app.worker_api.orchestrator.get_service_logs", return_value="line1\nline2") as mock_logs: + resp = _client().get("/api/containers/honeygain/logs", headers=_auth()) + assert resp.status_code == 200 + assert resp.json() == {"logs": "line1\nline2"} + mock_logs.assert_called_once_with("honeygain", lines=50) + + def test_logs_lines_capped_at_1000(self): + with patch("app.worker_api.orchestrator.get_service_logs", return_value="") as mock_logs: + resp = _client().get("/api/containers/honeygain/logs?lines=5000", headers=_auth()) + assert resp.status_code == 200 + mock_logs.assert_called_once_with("honeygain", lines=1000) + + def test_logs_not_found_returns_404(self): + with patch("app.worker_api.orchestrator.get_service_logs", side_effect=ValueError("gone")): + resp = _client().get("/api/containers/x/logs", headers=_auth()) + assert resp.status_code == 404 + + def test_logs_docker_unavailable_returns_503(self): + with patch("app.worker_api.orchestrator.get_service_logs", side_effect=RuntimeError("no docker")): + resp = _client().get("/api/containers/honeygain/logs", headers=_auth()) + assert resp.status_code == 503 From cd1c569f65e00cc13c0151f470a1e512587769a8 Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:35:09 +0200 Subject: [PATCH 6/6] feat(security): setup-token gate for first-run + trusted-proxy client IP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The "private network only" guard on first-run /register trusts request.client.host, which behind a reverse proxy is the proxy (a loopback/ private address) — so the check always passes and the first public visitor could seize the owner account. Add a proxy-independent gate: while no users exist, startup generates a one-time setup token, persists it (encrypted config), and logs it; /register then requires it (query param, X-Setup-Token header, or form field) and it is retired the moment the owner account is created. Also resolve the real client IP behind a trusted proxy (opt-in CASHPILOT_TRUSTED_PROXY, right-most X-Forwarded-For), used by both the first-run network check and the login rate limiter — which previously collapsed every proxied client into one bucket keyed by the proxy IP. Tests: token module + client_ip (trusted/untrusted) + the first-run guard (missing/wrong/query/header/form token, public-IP-still-blocked, network-only fallback once retired) + register integration (403 without token, 303 + token cleared with it); conftest resets the token global between tests. --- app/deps.py | 53 +++++++++++++++++-- app/main.py | 49 +++++++++++++++--- app/routers/auth.py | 18 +++++-- app/setup_token.py | 62 ++++++++++++++++++++++ app/templates/auth.html | 3 ++ tests/conftest.py | 15 ++++++ tests/test_main_routes.py | 52 +++++++++++++++++++ tests/test_setup_token.py | 105 ++++++++++++++++++++++++++++++++++++++ 8 files changed, 342 insertions(+), 15 deletions(-) create mode 100644 app/setup_token.py create mode 100644 tests/test_setup_token.py diff --git a/app/deps.py b/app/deps.py index d6bdc64..7fb4558 100644 --- a/app/deps.py +++ b/app/deps.py @@ -11,13 +11,14 @@ from __future__ import annotations import ipaddress +import os from typing import Any from fastapi import HTTPException, Request from fastapi.responses import RedirectResponse from fastapi.templating import Jinja2Templates -from app import auth +from app import auth, setup_token __all__ = [ "templates", @@ -26,8 +27,15 @@ "_require_writer", "_require_owner", "_require_private_network", + "_require_first_run_access", + "client_ip", ] +# Opt-in: set CASHPILOT_TRUSTED_PROXY=1 only when the app sits behind exactly one +# reverse proxy you control. X-Forwarded-For is attacker-controlled, so we ignore +# it unless the operator asserts a trusted proxy is stripping/appending it. +_TRUST_PROXY = os.getenv("CASHPILOT_TRUSTED_PROXY", "").strip().lower() in ("1", "true", "yes", "on") + templates = Jinja2Templates(directory="app/templates") @@ -57,13 +65,48 @@ def _require_owner(request: Request) -> dict[str, Any]: return user +def client_ip(request: Request) -> str | None: + """Best-effort real client IP. + + Behind a trusted reverse proxy (opt-in via ``CASHPILOT_TRUSTED_PROXY``) the + real peer is the right-most ``X-Forwarded-For`` entry — the value appended by + the trusted proxy, which a client cannot forge by prepending its own. Without + that opt-in we never trust the header and use the direct peer. + """ + if _TRUST_PROXY: + xff = request.headers.get("x-forwarded-for", "") + parts = [p.strip() for p in xff.split(",") if p.strip()] + if parts: + return parts[-1] + return request.client.host if request.client else None + + def _require_private_network(request: Request) -> None: - """Block requests from public IPs (for first-run setup).""" - if not request.client or not request.client.host: + """Block requests whose real client IP is public (first-run defense in depth).""" + ip_str = client_ip(request) + if not ip_str: return try: - client_ip = ipaddress.ip_address(request.client.host) + ip = ipaddress.ip_address(ip_str) except ValueError: return - if not (client_ip.is_loopback or client_ip.is_private): + if not (ip.is_loopback or ip.is_private): raise HTTPException(status_code=403, detail="First-run setup only allowed from private networks") + + +def _require_first_run_access(request: Request, setup_token_value: str | None = None) -> None: + """Gate first-run owner creation: private network AND the one-time setup token. + + The network check alone is spoofable behind a reverse proxy (the peer is then + the proxy), so the setup token — printed to the server logs, readable only + with host access — is the real gate. The token is accepted from the explicit + argument (a form field), the ``setup_token`` query param, or the + ``X-Setup-Token`` header. + """ + _require_private_network(request) + token = setup_token_value or request.query_params.get("setup_token") or request.headers.get("x-setup-token") + if not setup_token.verify(token): + raise HTTPException( + status_code=403, + detail="First-run setup requires the setup token printed in the server logs", + ) diff --git a/app/main.py b/app/main.py index e948b3a..cdeb292 100644 --- a/app/main.py +++ b/app/main.py @@ -31,7 +31,7 @@ from pydantic import BaseModel from starlette.middleware.base import BaseHTTPMiddleware -from app import auth, catalog, compose_generator, database, exchange_rates, fleet_key, metrics +from app import auth, catalog, compose_generator, database, exchange_rates, fleet_key, metrics, setup_token logging.basicConfig( level=logging.INFO, @@ -75,16 +75,16 @@ def _on_done(t: asyncio.Task) -> None: _LOGIN_WINDOW_SECONDS = 300 -def _check_login_rate(client_ip: str) -> None: +def _check_login_rate(ip: str) -> None: now = monotonic() - attempts = _login_attempts[client_ip] - _login_attempts[client_ip] = [t for t in attempts if now - t < _LOGIN_WINDOW_SECONDS] - if len(_login_attempts[client_ip]) >= _LOGIN_MAX_ATTEMPTS: + attempts = _login_attempts[ip] + _login_attempts[ip] = [t for t in attempts if now - t < _LOGIN_WINDOW_SECONDS] + if len(_login_attempts[ip]) >= _LOGIN_MAX_ATTEMPTS: raise HTTPException(status_code=429, detail="Too many login attempts. Try again in a few minutes.") -def _record_failed_login(client_ip: str) -> None: - _login_attempts[client_ip].append(monotonic()) +def _record_failed_login(ip: str) -> None: + _login_attempts[ip].append(monotonic()) def _safe_json(raw: str, fallback: Any = None) -> Any: @@ -273,6 +273,15 @@ async def _run_data_retention() -> None: logger.warning("Data retention error: %s", exc) +async def _run_vacuum() -> None: + """Reclaim disk left by retention deletes (SQLite does not auto-shrink).""" + try: + await database.vacuum_database() + logger.info("Database VACUUM complete") + except Exception as exc: + logger.warning("Database VACUUM error: %s", exc) + + async def _check_stale_workers() -> None: """Mark workers as offline if stale, and purge workers offline > 1 hour.""" try: @@ -312,6 +321,21 @@ async def lifespan(app: FastAPI): _changed = _u.get("password_changed_at") or 0.0 if _changed: auth.set_user_pwd_epoch(_u["id"], _changed) + # First-run setup token: while no users exist, require a one-time token + # (printed below) for /register so a proxy-exposed instance cannot be seized + # by the first public visitor. Persisted in config so it survives restarts; + # cleared once the owner account is created. + if not await database.has_any_users(): + _tok = await database.get_config("_setup_token") + if not _tok: + _tok = setup_token.generate() + await database.set_config("_setup_token", _tok) + setup_token.set_active(_tok) + logger.warning( + "FIRST-RUN SETUP: no account exists yet. Create the owner account at " + "/register?setup_token=%s — this token is required and is shown only here.", + _tok, + ) catalog.load_services() catalog.register_sighup() @@ -355,6 +379,15 @@ def _on_job_event(event): coalesce=True, misfire_grace_time=300, ) + scheduler.add_job( + _run_vacuum, + "interval", + weeks=1, + id="db_vacuum", + max_instances=1, + coalesce=True, + misfire_grace_time=300, + ) scheduler.add_job( exchange_rates.refresh, "interval", @@ -424,9 +457,11 @@ async def dispatch(self, request, call_next): from app.deps import ( # noqa: E402 _login_redirect, # noqa: F401 (re-exported for app.main.* test/router surface) _require_auth_api, + _require_first_run_access, # noqa: F401 (re-exported for app.main.* router surface) _require_owner, _require_private_network, # noqa: F401 (re-exported for app.main.* router surface) _require_writer, + client_ip, # noqa: F401 (re-exported for app.main.* router surface) templates, # noqa: F401 (re-exported for app.main.* router/test surface) ) diff --git a/app/routers/auth.py b/app/routers/auth.py index aa8ff62..fe53613 100644 --- a/app/routers/auth.py +++ b/app/routers/auth.py @@ -45,7 +45,7 @@ async def do_login( username: str = Form(...), password: str = Form(...), ): - client_ip = request.client.host if request.client else "unknown" + client_ip = main.client_ip(request) or "unknown" try: main._check_login_rate(client_ip) except HTTPException: @@ -86,7 +86,7 @@ async def page_register(request: Request, error: str = ""): if not user or user.get("r") != "owner": return RedirectResponse("/login", status_code=303) if is_first: - main._require_private_network(request) + main._require_first_run_access(request) return main.templates.TemplateResponse( request, @@ -99,6 +99,8 @@ async def page_register(request: Request, error: str = ""): "button_text": "Create Account", "error": error, "is_first": is_first, + # Carry the setup token through the form so the POST passes the same gate. + "setup_token": request.query_params.get("setup_token", "") if is_first else "", }, ) @@ -109,6 +111,7 @@ async def do_register( username: str = Form(...), password: str = Form(...), password_confirm: str = Form(...), + setup_token: str = Form(""), ): is_first = not await main.database.has_any_users() @@ -119,7 +122,7 @@ async def do_register( raise HTTPException(status_code=403, detail="Only owners can add users") if is_first: - main._require_private_network(request) + main._require_first_run_access(request, setup_token) if not re.match(r"^[a-zA-Z0-9_-]{3,32}$", username): return main.templates.TemplateResponse( @@ -133,6 +136,7 @@ async def do_register( "button_text": "Create Account", "error": "Username must be 3-32 alphanumeric characters (a-z, 0-9, _ -)", "is_first": is_first, + "setup_token": setup_token if is_first else "", }, status_code=400, ) @@ -149,6 +153,7 @@ async def do_register( "button_text": "Create Account", "error": "Passwords do not match", "is_first": is_first, + "setup_token": setup_token if is_first else "", }, status_code=400, ) @@ -165,6 +170,7 @@ async def do_register( "button_text": "Create Account", "error": "Password must be at least 10 characters", "is_first": is_first, + "setup_token": setup_token if is_first else "", }, status_code=400, ) @@ -182,6 +188,7 @@ async def do_register( "button_text": "Create Account", "error": "Username already taken", "is_first": is_first, + "setup_token": setup_token if is_first else "", }, status_code=400, ) @@ -191,6 +198,11 @@ async def do_register( hashed = main.auth.hash_password(password) user_id = await main.database.create_user(username, hashed, role) + if is_first: + # Owner now exists — retire the one-time setup token permanently. + await main.database.delete_config_keys(["_setup_token"]) + main.setup_token.clear() + token = main.auth.create_session_token(user_id, username, role) dest = "/setup" if is_first else "/" response = RedirectResponse(dest, status_code=303) diff --git a/app/setup_token.py b/app/setup_token.py new file mode 100644 index 0000000..cf95235 --- /dev/null +++ b/app/setup_token.py @@ -0,0 +1,62 @@ +"""First-run setup-token gate. + +The very first account created on a fresh install becomes the ``owner``. The +private-network check in :mod:`app.deps` is not enough on its own: behind a +reverse proxy ``request.client`` is the *proxy* (a loopback/private address), so +the network check always passes and the first person to reach ``/register`` from +the public internet could seize the owner account. + +This module adds a proxy-independent second factor: a one-time token generated at +startup while no users exist, printed to the container logs. Only someone who can +read the server logs (i.e. already has host access) can complete first-run setup. +Once the owner account is created the token is cleared and never required again +(further users are added by an authenticated owner). + +The active token is held in a module global so the synchronous request guard can +check it without a DB round-trip; it is also persisted in the ``config`` table so +it survives restarts until consumed. +""" + +from __future__ import annotations + +import hmac +import secrets + +# The token currently required for first-run registration, or ``None`` when no +# first-run gate is active (owner already exists, or not yet initialised). +_active: str | None = None + + +def generate() -> str: + """Return a fresh, URL-safe setup token.""" + return secrets.token_urlsafe(24) + + +def set_active(token: str | None) -> None: + """Install (or clear, with ``None``) the token required for first-run setup.""" + global _active + _active = token or None + + +def clear() -> None: + """Drop the first-run gate — called once the owner account exists.""" + set_active(None) + + +def active() -> str | None: + """Return the currently required setup token, or ``None`` if none is active.""" + return _active + + +def verify(provided: str | None) -> bool: + """Check a caller-supplied token against the active one. + + Returns ``True`` when no token is active (nothing to enforce) or when the + supplied value matches in constant time; ``False`` otherwise. + """ + current = _active + if current is None: + return True + if not provided: + return False + return hmac.compare_digest(current, provided) diff --git a/app/templates/auth.html b/app/templates/auth.html index c3b5da8..e457129 100644 --- a/app/templates/auth.html +++ b/app/templates/auth.html @@ -151,6 +151,9 @@ {% endif %}
+ {% if mode == "register" and is_first %} + + {% endif %}
diff --git a/tests/conftest.py b/tests/conftest.py index 1c86c0f..1ee50b4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -62,3 +62,18 @@ def _reset_login_attempts(): main._login_attempts.clear() yield + + +@pytest.fixture(autouse=True) +def _reset_setup_token(): + """Clear the first-run setup-token module global before every test. + + ``app.setup_token._active`` persists for the whole process; a test that runs + lifespan on a fresh DB (or exercises the token directly) would otherwise leak + an active token into later tests, making unrelated /register tests 403. + """ + with contextlib.suppress(Exception): + from app import setup_token + + setup_token.clear() + yield diff --git a/tests/test_main_routes.py b/tests/test_main_routes.py index c66de19..4e082bb 100644 --- a/tests/test_main_routes.py +++ b/tests/test_main_routes.py @@ -264,6 +264,53 @@ def test_register_first_user_success(self, client): patch("app.main.database.get_user_by_username", new_callable=AsyncMock, return_value=None), patch("app.main.auth.hash_password", return_value="hashed"), patch("app.main.database.create_user", new_callable=AsyncMock, return_value=1), + patch("app.main.database.delete_config_keys", new_callable=AsyncMock), + patch("app.main.auth.create_session_token", return_value="tok"), + patch("app.main.auth.set_session_cookie", side_effect=lambda r, t: r), + ): + resp = client.post( + "/register", + data={ + "username": "admin", + "password": "password123", + "password_confirm": "password123", + }, + follow_redirects=False, + ) + assert resp.status_code == 303 + + def test_register_first_user_requires_setup_token(self, client): + # When a first-run setup token is active, registering without it is refused + # (the proxy-independent gate against a public visitor seizing the owner). + from app import setup_token + + setup_token.set_active("the-token") + with ( + _no_auth(), + patch("app.main.database.has_any_users", new_callable=AsyncMock, return_value=False), + ): + resp = client.post( + "/register", + data={ + "username": "admin", + "password": "password123", + "password_confirm": "password123", + }, + follow_redirects=False, + ) + assert resp.status_code == 403 + + def test_register_first_user_with_setup_token_clears_it(self, client): + from app import setup_token + + setup_token.set_active("the-token") + with ( + _no_auth(), + patch("app.main.database.has_any_users", new_callable=AsyncMock, return_value=False), + patch("app.main.database.get_user_by_username", new_callable=AsyncMock, return_value=None), + patch("app.main.auth.hash_password", return_value="hashed"), + patch("app.main.database.create_user", new_callable=AsyncMock, return_value=1), + patch("app.main.database.delete_config_keys", new_callable=AsyncMock) as del_keys, patch("app.main.auth.create_session_token", return_value="tok"), patch("app.main.auth.set_session_cookie", side_effect=lambda r, t: r), ): @@ -273,10 +320,13 @@ def test_register_first_user_success(self, client): "username": "admin", "password": "password123", "password_confirm": "password123", + "setup_token": "the-token", }, follow_redirects=False, ) assert resp.status_code == 303 + del_keys.assert_awaited_once_with(["_setup_token"]) + assert setup_token.active() is None def test_register_password_mismatch(self, client): with ( @@ -1791,6 +1841,7 @@ async def _run(): patch("app.main.database.connect_shared", new_callable=AsyncMock), patch("app.main.database.close_shared", new_callable=AsyncMock), patch("app.main.database.list_users_with_pwd_epoch", new_callable=AsyncMock, return_value=[]), + patch("app.main.database.has_any_users", new_callable=AsyncMock, return_value=True), patch("app.main.catalog.load_services"), patch("app.main.catalog.register_sighup"), patch("app.main.exchange_rates.refresh", new_callable=AsyncMock), @@ -1831,6 +1882,7 @@ async def _run(): patch("app.main.database.connect_shared", new_callable=AsyncMock), patch("app.main.database.close_shared", new_callable=AsyncMock), patch("app.main.database.list_users_with_pwd_epoch", new_callable=AsyncMock, return_value=[]), + patch("app.main.database.has_any_users", new_callable=AsyncMock, return_value=True), patch("app.main.catalog.load_services"), patch("app.main.catalog.register_sighup"), patch("app.main.exchange_rates.refresh", new_callable=AsyncMock), diff --git a/tests/test_setup_token.py b/tests/test_setup_token.py new file mode 100644 index 0000000..59621e5 --- /dev/null +++ b/tests/test_setup_token.py @@ -0,0 +1,105 @@ +"""Tests for the first-run setup-token gate (app/setup_token.py + app/deps).""" + +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import HTTPException + +from app import deps, setup_token + + +def _req(host="127.0.0.1", xff=None, query=None, headers=None): + """A minimal fake Request: real dicts for headers/query_params so .get works.""" + req = MagicMock() + if host is None: + req.client = None + else: + req.client.host = host + h = {} + if xff is not None: + h["x-forwarded-for"] = xff + if headers: + h.update(headers) + req.headers = h + req.query_params = query or {} + return req + + +class TestSetupTokenModule: + def test_verify_no_active_allows_anything(self): + setup_token.clear() + assert setup_token.verify(None) is True + assert setup_token.verify("whatever") is True + + def test_verify_active_requires_exact_match(self): + setup_token.set_active("s3cret") + assert setup_token.verify("s3cret") is True + assert setup_token.verify("wrong") is False + assert setup_token.verify(None) is False + assert setup_token.verify("") is False + + def test_generate_is_unique_and_long(self): + a, b = setup_token.generate(), setup_token.generate() + assert a != b + assert len(a) >= 20 + + def test_set_active_empty_string_deactivates(self): + setup_token.set_active("") + assert setup_token.active() is None + + +class TestClientIp: + def test_ignores_xff_without_trusted_proxy(self): + with patch.object(deps, "_TRUST_PROXY", False): + assert deps.client_ip(_req("10.0.0.1", xff="1.2.3.4")) == "10.0.0.1" + + def test_uses_rightmost_xff_with_trusted_proxy(self): + # The trusted proxy appends the real peer on the right; a client can only + # forge entries on the left, so the right-most value is authoritative. + with patch.object(deps, "_TRUST_PROXY", True): + req = _req("10.0.0.1", xff="9.9.9.9, 8.8.8.8, 203.0.113.5") + assert deps.client_ip(req) == "203.0.113.5" + + def test_trusted_proxy_falls_back_to_peer_without_xff(self): + with patch.object(deps, "_TRUST_PROXY", True): + assert deps.client_ip(_req("10.0.0.1")) == "10.0.0.1" + + def test_no_client_returns_none(self): + with patch.object(deps, "_TRUST_PROXY", False): + assert deps.client_ip(_req(host=None)) is None + + +class TestRequireFirstRunAccess: + def test_blocks_when_token_active_and_missing(self): + setup_token.set_active("tok") + with patch.object(deps, "_TRUST_PROXY", False), pytest.raises(HTTPException) as ei: + deps._require_first_run_access(_req("127.0.0.1")) + assert ei.value.status_code == 403 + + def test_allows_with_correct_token_arg(self): + setup_token.set_active("tok") + with patch.object(deps, "_TRUST_PROXY", False): + assert deps._require_first_run_access(_req("127.0.0.1"), "tok") is None + + def test_token_from_query_param(self): + setup_token.set_active("tok") + with patch.object(deps, "_TRUST_PROXY", False): + assert deps._require_first_run_access(_req("127.0.0.1", query={"setup_token": "tok"})) is None + + def test_token_from_header(self): + setup_token.set_active("tok") + with patch.object(deps, "_TRUST_PROXY", False): + assert deps._require_first_run_access(_req("127.0.0.1", headers={"x-setup-token": "tok"})) is None + + def test_public_ip_blocked_even_with_correct_token(self): + # The network check runs first: a public client is refused regardless of token. + setup_token.set_active("tok") + with patch.object(deps, "_TRUST_PROXY", False), pytest.raises(HTTPException) as ei: + deps._require_first_run_access(_req("8.8.8.8"), "tok") + assert ei.value.status_code == 403 + + def test_no_active_token_falls_back_to_network_only(self): + # Once the token is retired, first-run access is network-gated only. + setup_token.clear() + with patch.object(deps, "_TRUST_PROXY", False): + assert deps._require_first_run_access(_req("192.168.1.5")) is None