fix: ecosystem audit — auth hardening, compose fixes, fleet docs, XSS, validation (#46)

* fix: address ecosystem audit — security, compose, docs, validation

Security:
- Fleet key now grants limited 'fleet' role instead of full 'writer'
- Deploy endpoint requires owner role (not writer/fleet)
- First-run registration restricted to private/loopback networks
- XSS fix: fleet page uses data attributes instead of inline onclick
- Fleet bootstrap: entrypoint ensures /fleet dir is writable

Compose generator:
- Named volumes now declared at top-level volumes section
- ${VAR} placeholders escaped as $${VAR} to prevent host interpolation

Validation:
- Deploy API rejects requests with blank required env vars
- Android workers: containers array skipped (only apps processed)

Docs:
- Fleet docs rewritten to match actual REST heartbeat implementation
- Presearch/IPRoyal/Earn.fm/Repocket guide env vars aligned with YAML
- Earn.fm/Repocket service YAMLs aligned with collector auth models

Testing:
- Added requirements-dev.txt for local test reproducibility

Closes #cp-bj8p #cp-jbkg #cp-370w #cp-3ddy #cp-zvhs #cp-xt5k #cp-zrlr
Closes #cp-pkio #cp-v65p #cp-p2sb #cp-hvay #cp-6hr9 #cp-6ho7

* fix: address review findings — extract IP helper, strip whitespace, fix SSRF, align docs

- Extract _require_private_network() helper (removes duplication, handles None/ValueError)
- Strip whitespace in required env var validation
- Fix SSRF in api_worker_command by using _get_verified_worker_url()
- Align earnfm/repocket docs with email/password auth model
- Fix repocket policy contradiction (VPS accepted at lower rates)

* fix: address all review nits — escape idempotency, NaN guard, test coverage

- Fix _escape_interpolation to use regex (skips already-escaped $${})
- Add isNaN guard in fleet.html event delegation
- Add TestRequireRole covering fleet escalation boundaries
- Add TestEscapeInterpolation, TestIsNamedVolume, TestNamedVolumeDeclaration

* fix: update tests for auth changes — deploy requires owner, fleet returns fleet role

- Fix _require_private_network to allow non-IP hosts (test clients, unix sockets)
- Update deploy tests to use owner auth (deploy endpoint now requires owner)
- Update auth test to expect fleet role instead of writer
- Fix ruff format on compose_generator.py and test_compose.py

* style: split long comprehension lines for format compatibility

Author: GeiserX

app/auth.py

@@ -104,15 +104,15 @@ def get_current_user(request: Request) -> dict[str, Any] | None:
     Checks Authorization header first (for programmatic access like Home Assistant),
     then falls back to session cookie (for browser sessions).
     """
-    # Check Bearer token — admin key gets owner, fleet key gets writer
+    # Check Bearer token — admin key gets owner, fleet key gets fleet (limited)
     auth_header = request.headers.get("Authorization", "")
     if auth_header:
         admin_key = os.getenv("CASHPILOT_ADMIN_API_KEY", "")
         if admin_key and auth_header == f"Bearer {admin_key}":
             return {"uid": 0, "u": "api", "r": "owner"}
         resolved_fleet_key = _fleet_key_mod.resolve_fleet_key()
         if resolved_fleet_key and auth_header == f"Bearer {resolved_fleet_key}":
-            return {"uid": 0, "u": "api", "r": "writer"}
+            return {"uid": 0, "u": "fleet", "r": "fleet"}
 
     # Fall back to session cookie
     token = request.cookies.get(SESSION_COOKIE)
@@ -148,4 +148,8 @@ def require_role(user: dict[str, Any] | None, *roles: str) -> bool:
     """Check if user has one of the required roles."""
     if not user:
         return False
-    return user.get("r") in roles
+    role = user.get("r")
+    # fleet role implicitly satisfies writer checks (for heartbeat/status endpoints)
+    if role == "fleet" and "writer" in roles:
+        return True
+    return role in roles

app/compose_generator.py

@@ -11,6 +11,7 @@
 
 from __future__ import annotations
 
+import re
 import socket
 from typing import Any
 
@@ -27,6 +28,26 @@
 )
 
 
+def _escape_interpolation(value: str) -> str:
+    """Escape ${VAR} as $${VAR} so Docker Compose treats it as literal.
+
+    Skips already-escaped sequences ($${). Does not handle bare $VAR (without braces)
+    since catalog YAML exclusively uses the braced form.
+    """
+    return re.sub(r"(?<!\$)\$\{", "$${", value)
+
+
+def _is_named_volume(volume_str: str) -> str | None:
+    """Return the volume name if the mapping uses a named volume, else None.
+
+    Named volumes have a source that doesn't start with /, ., or ~.
+    """
+    source = volume_str.split(":")[0]
+    if source and not source.startswith(("/", ".", "~")):
+        return source
+    return None
+
+
 def _service_to_compose(
     svc: dict[str, Any],
     env_vars: dict[str, str] | None = None,
@@ -87,10 +108,10 @@ def _service_to_compose(
     if ports:
         compose_svc["ports"] = [str(p) for p in ports]
 
-    # Volumes
+    # Volumes — escape ${VAR} interpolation in host paths
     volumes = docker_conf.get("volumes", [])
     if volumes:
-        compose_svc["volumes"] = [str(v) for v in volumes]
+        compose_svc["volumes"] = [_escape_interpolation(str(v)) for v in volumes]
 
     # Network mode
     network_mode = docker_conf.get("network_mode")
@@ -102,10 +123,10 @@ def _service_to_compose(
     if cap_add:
         compose_svc["cap_add"] = cap_add
 
-    # Command
+    # Command — escape ${VAR} interpolation
     command = docker_conf.get("command")
     if command:
-        compose_svc["command"] = command
+        compose_svc["command"] = _escape_interpolation(command)
 
     return compose_svc
 
@@ -169,6 +190,17 @@ def generate_compose_all(
 
 def _dump_compose(compose: dict, name: str) -> str:
     """Serialize compose dict to YAML with a header comment."""
+    # Collect named volumes from all services
+    named_volumes: set[str] = set()
+    for svc in compose.get("services", {}).values():
+        for vol in svc.get("volumes", []):
+            vol_name = _is_named_volume(vol)
+            if vol_name:
+                named_volumes.add(vol_name)
+
+    if named_volumes:
+        compose["volumes"] = {v: {} for v in sorted(named_volumes)}
+
     header = (
         f"# Generated by CashPilot for: {name}\n"
         "# https://github.com/GeiserX/CashPilot\n"

app/fleet_key.py

@@ -62,8 +62,10 @@ def resolve_fleet_key() -> str:
                 pass
             time.sleep(0.1)
     except OSError as exc:
-        _logger.warning(
-            "Could not persist fleet key to %s: %s — set CASHPILOT_API_KEY or mount a shared /fleet volume",
+        _logger.error(
+            "Cannot write fleet key to %s: %s. "
+            "Fix: set CASHPILOT_API_KEY env var explicitly, "
+            "or fix /fleet volume permissions (chown 1000:0 /fleet on the host).",
             _FLEET_KEY_FILE,
             exc,
         )

app/main.py

@@ -62,27 +62,28 @@ async def _get_all_worker_containers() -> list[dict[str, Any]]:
         is_android = sys_info.get("device_type") == "android"
         worker_name = w.get("name", "worker")
 
-        # Docker containers (from Docker-based workers)
-        containers = _safe_json(w.get("containers", "[]"))
-        for c in containers:
-            slug = c.get("slug", "")
-            if slug:
-                result.append(
-                    {
-                        "slug": slug,
-                        "name": c.get("name", slug),
-                        "status": c.get("status", "unknown"),
-                        "image": c.get("image", ""),
-                        "cpu_percent": c.get("cpu_percent", 0),
-                        "memory_mb": c.get("memory_mb", 0),
-                        "category": "",
-                        "deployed_by": worker_name,
-                        "_node": worker_name,
-                        "_worker_id": w.get("id"),
-                        "_has_docker": worker_has_docker,
-                        "_is_android": False,
-                    }
-                )
+        # Docker containers (from Docker-based workers only — skip for Android)
+        if not is_android:
+            containers = _safe_json(w.get("containers", "[]"))
+            for c in containers:
+                slug = c.get("slug", "")
+                if slug:
+                    result.append(
+                        {
+                            "slug": slug,
+                            "name": c.get("name", slug),
+                            "status": c.get("status", "unknown"),
+                            "image": c.get("image", ""),
+                            "cpu_percent": c.get("cpu_percent", 0),
+                            "memory_mb": c.get("memory_mb", 0),
+                            "category": "",
+                            "deployed_by": worker_name,
+                            "_node": worker_name,
+                            "_worker_id": w.get("id"),
+                            "_has_docker": worker_has_docker,
+                            "_is_android": False,
+                        }
+                    )
 
         # Android apps (from Android workers)
         if is_android:
@@ -309,6 +310,18 @@ def _require_owner(request: Request) -> dict[str, Any]:
     return user
 
 
+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:
+        return
+    try:
+        client_ip = ipaddress.ip_address(request.client.host)
+    except ValueError:
+        return
+    if not (client_ip.is_loopback or client_ip.is_private):
+        raise HTTPException(status_code=403, detail="First-run setup only allowed from private networks")
+
+
 # ---------------------------------------------------------------------------
 # Auth routes
 # ---------------------------------------------------------------------------
@@ -373,6 +386,8 @@ async def page_register(request: Request, error: str = ""):
         user = auth.get_current_user(request)
         if not user or user.get("r") != "owner":
             return RedirectResponse("/login", status_code=303)
+    if is_first:
+        _require_private_network(request)
 
     return templates.TemplateResponse(
         request,
@@ -404,6 +419,9 @@ async def do_register(
         if not user or user.get("r") != "owner":
             raise HTTPException(status_code=403, detail="Only owners can add users")
 
+    if is_first:
+        _require_private_network(request)
+
     if not re.match(r"^[a-zA-Z0-9_-]{3,32}$", username):
         return templates.TemplateResponse(
             request,
@@ -773,7 +791,7 @@ class DeployRequest(BaseModel):
 
 @app.post("/api/deploy/{slug}")
 async def api_deploy(request: Request, slug: str, body: DeployRequest, worker_id: int | None = None) -> dict[str, str]:
-    _require_writer(request)
+    _require_owner(request)
     worker_id = await _resolve_worker_id(worker_id)
     svc = catalog.get_service(slug)
     if not svc:
@@ -794,6 +812,15 @@ async def api_deploy(request: Request, slug: str, body: DeployRequest, worker_id
         env[var["key"]] = str(default)
     env.update(body.env or {})
 
+    # Validate required env vars are not blank
+    missing = [
+        var.get("label", var["key"])
+        for var in docker_conf.get("env", [])
+        if var.get("required") and not env.get(var["key"], "").strip()
+    ]
+    if missing:
+        raise HTTPException(status_code=400, detail=f"Missing required fields: {', '.join(missing)}")
+
     # Ports — key is "container_port/protocol" per Docker SDK
     ports: dict[str, int] = {}
     for mapping in docker_conf.get("ports", []):
@@ -1651,17 +1678,7 @@ async def api_worker_command(request: Request, worker_id: int, body: WorkerComma
     _require_writer(request)
 
     worker = await database.get_worker(worker_id)
-    if not worker:
-        raise HTTPException(status_code=404, detail="Worker not found")
-    if worker["status"] != "online":
-        raise HTTPException(status_code=503, detail="Worker is offline")
-    if not worker["url"]:
-        raise HTTPException(status_code=503, detail="Worker URL not known")
-
-    url = worker["url"].rstrip("/")
-    headers = {}
-    if FLEET_API_KEY:
-        headers["Authorization"] = f"Bearer {FLEET_API_KEY}"
+    url, headers = _get_verified_worker_url(worker)
 
     try:
         async with httpx.AsyncClient(timeout=30) as client:

app/templates/fleet.html

@@ -122,7 +122,7 @@ <h2 class="section-title">Workers</h2>
                 <span style="font-weight: 600; color: var(--text-primary);">${esc(w.name)}</span>
                 ${isAndroid ? '<span style="font-size:0.7rem; padding:2px 6px; border-radius:4px; background:var(--bg-hover); color:var(--text-muted);">Android</span>' : ''}
               </div>
-              ${_isOwner ? `<button class="btn btn-danger btn-sm" onclick="removeWorker(${w.id}, '${esc(w.name)}')" title="Remove worker">Remove</button>` : ''}
+              ${_isOwner ? `<button class="btn btn-danger btn-sm btn-remove-worker" data-worker-id="${w.id}" data-worker-name="${esc(w.name)}" title="Remove worker">Remove</button>` : ''}
             </div>
             <div style="display: flex; gap: 16px; flex-wrap: wrap; font-size: 0.82rem; color: var(--text-muted);">
               <span>${esc(w.url || '-')}</span>
@@ -202,7 +202,7 @@ <h2 class="section-title">Workers</h2>
     }
   };
 
-  window.removeWorker = async function(workerId, name) {
+  async function removeWorker(workerId, name) {
     if (!confirm(`Remove worker "${name}"? This will unregister it from the fleet.`)) return;
     try {
       await CP.api(`/api/workers/${workerId}`, { method: 'DELETE' });
@@ -211,7 +211,16 @@ <h2 class="section-title">Workers</h2>
     } catch (err) {
       CP.toast('Failed: ' + err.message, 'error');
     }
-  };
+  }
+
+  document.getElementById('fleet-workers').addEventListener('click', function(e) {
+    const btn = e.target.closest('.btn-remove-worker');
+    if (!btn) return;
+    const id = parseInt(btn.dataset.workerId, 10);
+    if (isNaN(id)) return;
+    const name = btn.dataset.workerName;
+    removeWorker(id, name);
+  });
 
   window.copyWorkerEnv = async function() {
     // Auto-fetch API key if not yet revealed