From 572b5f2dcb824519ddffee818e64980bc95df55d Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 14:28:05 +0200
Subject: [PATCH 1/9] feat: traffmonetizer browser JWT auth + volume cleanup on
 remove
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Traffmonetizer now enforces reCAPTCHA on login, permanently blocking
programmatic email/password auth. Switch collector to browser-extracted
JWT (same pattern as bytelixir/grass). Simpler implementation: token-only
constructor, clear error messages guiding users to Local Storage.

Add delete_volumes flag to remove_service that cleans up named Docker
volumes when removing a container. Flows end-to-end: UI → main API →
worker API → orchestrator. Bind mounts and anonymous volumes are skipped.
---
 app/collectors/traffmonetizer.py      | 67 ++++++++-------------------
 app/main.py                           | 22 ++++++---
 app/orchestrator.py                   | 39 ++++++++++++++--
 app/worker_api.py                     |  8 ++--
 services/bandwidth/traffmonetizer.yml |  3 +-
 tests/test_collectors_deep.py         | 38 ++++++---------
 tests/test_coverage_gaps.py           | 12 +++--
 7 files changed, 98 insertions(+), 91 deletions(-)

diff --git a/app/collectors/traffmonetizer.py b/app/collectors/traffmonetizer.py
index d33c570..0db47eb 100644
--- a/app/collectors/traffmonetizer.py
+++ b/app/collectors/traffmonetizer.py
@@ -1,7 +1,12 @@
 """Traffmonetizer earnings collector.
 
-Uses the Traffmonetizer dashboard API with token-based authentication
-to fetch device statistics and earnings balance.
+Traffmonetizer enforces reCAPTCHA on their login endpoint, making
+programmatic email/password authentication impossible. Users must
+extract a JWT from their browser session.
+
+To get the token: open app.traffmonetizer.com, log in, press F12,
+go to Application > Local Storage > https://app.traffmonetizer.com,
+and copy the `access_token` value (a long JWT string).
 """
 
 from __future__ import annotations
@@ -18,70 +23,36 @@
 
 
 class TraffmonetizerCollector(BaseCollector):
-    """Collect earnings from Traffmonetizer's API."""
+    """Collect earnings from Traffmonetizer's API using a browser JWT."""
 
     platform = "traffmonetizer"
 
-    def __init__(self, email: str = "", password: str = "", token: str = "") -> None:
-        self.email = email
-        self.password = password
-        self._token: str | None = token or None
-
-    async def _authenticate(self, client: httpx.AsyncClient) -> str:
-        """Obtain a JWT token via email/password login."""
-        resp = await client.post(
-            f"{API_BASE}/auth/login",
-            json={
-                "email": self.email,
-                "password": self.password,
-                "g-recaptcha-response": "",
-            },
-            headers={"Origin": "https://app.traffmonetizer.com"},
-        )
-        resp.raise_for_status()
-        data = resp.json()
-        token = data.get("data", {}).get("token", "") or data.get("token", "")
-        if not token:
-            raise ValueError("No token in Traffmonetizer login response")
-        return token
+    def __init__(self, token: str = "") -> None:
+        self._token = token.strip()
 
     async def collect(self) -> EarningsResult:
         """Fetch current Traffmonetizer balance."""
+        if not self._token:
+            return EarningsResult(
+                platform=self.platform,
+                balance=0.0,
+                error="No token configured — extract access_token from browser Local Storage",
+            )
+
         try:
             async with httpx.AsyncClient(timeout=30) as client:
-                if not self._token and self.email:
-                    self._token = await self._authenticate(client)
-
-                if not self._token:
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="No token or credentials configured",
-                    )
-
-                headers = {
-                    "Authorization": f"Bearer {self._token}",
-                }
+                headers = {"Authorization": f"Bearer {self._token}"}
 
                 resp = await client.get(
                     f"{API_BASE}/app_user/get_balance",
                     headers=headers,
                 )
 
-                # Token may have expired — retry once with re-auth
-                if resp.status_code in (401, 403) and self.email:
-                    self._token = await self._authenticate(client)
-                    headers["Authorization"] = f"Bearer {self._token}"
-                    resp = await client.get(
-                        f"{API_BASE}/app_user/get_balance",
-                        headers=headers,
-                    )
-
                 if resp.status_code in (401, 403):
                     return EarningsResult(
                         platform=self.platform,
                         balance=0.0,
-                        error="Authentication failed — check credentials or token",
+                        error="Token expired — refresh access_token from browser Local Storage",
                     )
 
                 resp.raise_for_status()
diff --git a/app/main.py b/app/main.py
index 2025915..8f68592 100644
--- a/app/main.py
+++ b/app/main.py
@@ -885,10 +885,13 @@ async def api_restart(request: Request, slug: str, worker_id: int | None = None)
 
 
 @app.delete("/api/remove/{slug}")
-async def api_remove(request: Request, slug: str, worker_id: int | None = None) -> dict[str, str]:
+async def api_remove(
+    request: Request, slug: str, worker_id: int | None = None, delete_volumes: bool = False
+) -> dict[str, Any]:
     _require_writer(request)
     worker_id = await _resolve_worker_id(worker_id)
-    result = await _proxy_worker_command(worker_id, "remove", slug)
+    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)
     return result
 
@@ -934,7 +937,9 @@ def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[str, str
     return url, headers
 
 
-async def _proxy_worker_command(worker_id: int, command: str, slug: str) -> dict[str, str]:
+async def _proxy_worker_command(
+    worker_id: int, command: str, slug: str, *, params: dict[str, str] | None = None
+) -> 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)
@@ -942,7 +947,9 @@ async def _proxy_worker_command(worker_id: int, command: str, slug: str) -> dict
     try:
         async with httpx.AsyncClient(timeout=30) as client:
             if command == "remove":
-                resp = await client.delete(f"{url}/api/containers/{slug}", headers=headers)
+                resp = await client.delete(
+                    f"{url}/api/containers/{slug}", headers=headers, params=params
+                )
             else:
                 resp = await client.post(f"{url}/api/containers/{slug}/{command}", headers=headers)
             if resp.status_code >= 400:
@@ -1043,10 +1050,13 @@ async def api_service_logs(
 
 
 @app.delete("/api/services/{slug}")
-async def api_service_remove(request: Request, slug: str, worker_id: int | None = None) -> dict[str, str]:
+async def api_service_remove(
+    request: Request, slug: str, worker_id: int | None = None, delete_volumes: bool = False
+) -> dict[str, Any]:
     _require_writer(request)
     worker_id = await _resolve_worker_id(worker_id)
-    result = await _proxy_worker_command(worker_id, "remove", slug)
+    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)
     return result
 
diff --git a/app/orchestrator.py b/app/orchestrator.py
index afcf583..1bbe6ff 100644
--- a/app/orchestrator.py
+++ b/app/orchestrator.py
@@ -331,11 +331,44 @@ def restart_service(slug: str) -> None:
     logger.info("Restarted container %s", container.name)
 
 
-def remove_service(slug: str) -> None:
-    """Stop and remove the container for a service."""
+def remove_service(slug: str, delete_volumes: bool = False) -> dict[str, Any]:
+    """Stop and remove the container for a service.
+
+    When delete_volumes is True, also remove named volumes that were
+    mounted into the container. Bind mounts and anonymous volumes are skipped.
+    """
     container = _find_container(slug)
+    name = container.name
+
+    volume_names: list[str] = []
+    if delete_volumes:
+        for m in container.attrs.get("Mounts", []) or []:
+            if m.get("Type") == "volume" and m.get("Name"):
+                volume_names.append(m["Name"])
+
     container.remove(force=True)
-    logger.info("Removed container %s", container.name)
+    logger.info("Removed container %s", name)
+
+    deleted_volumes: list[str] = []
+    failed_volumes: list[str] = []
+    if volume_names:
+        client = _get_client()
+        for vol_name in volume_names:
+            try:
+                client.volumes.get(vol_name).remove(force=True)
+                deleted_volumes.append(vol_name)
+                logger.info("Removed volume %s", vol_name)
+            except NotFound:
+                deleted_volumes.append(vol_name)
+            except APIError as exc:
+                failed_volumes.append(vol_name)
+                logger.warning("Failed to remove volume %s: %s", vol_name, exc)
+
+    return {
+        "container": name,
+        "deleted_volumes": deleted_volumes,
+        "failed_volumes": failed_volumes,
+    }
 
 
 def start_service(slug: str) -> None:
diff --git a/app/worker_api.py b/app/worker_api.py
index 4192597..bd00877 100644
--- a/app/worker_api.py
+++ b/app/worker_api.py
@@ -354,11 +354,13 @@ async def api_start_container(request: Request, slug: str) -> dict[str, str]:
 
 
 @app.delete("/api/containers/{slug}")
-async def api_remove_container(request: Request, slug: str) -> dict[str, str]:
+async def api_remove_container(
+    request: Request, slug: str, delete_volumes: bool = False
+) -> dict[str, Any]:
     _verify_api_key(request)
     try:
-        orchestrator.remove_service(slug)
-        return {"status": "removed"}
+        result = orchestrator.remove_service(slug, delete_volumes=delete_volumes)
+        return {"status": "removed", **result}
     except ValueError as exc:
         raise HTTPException(status_code=404, detail=str(exc))
     except RuntimeError as exc:
diff --git a/services/bandwidth/traffmonetizer.yml b/services/bandwidth/traffmonetizer.yml
index 81e77d9..ed8635a 100644
--- a/services/bandwidth/traffmonetizer.yml
+++ b/services/bandwidth/traffmonetizer.yml
@@ -68,4 +68,5 @@ platforms: [docker, windows, macos, android]
 
 collector:
   type: api
-  notes: "API at api.traffmonetizer.com/api/dashboard. Bearer token auth. Returns balance in USD."
+  auth: browser_jwt
+  notes: "reCAPTCHA blocks programmatic login. Extract access_token from browser Local Storage at app.traffmonetizer.com."
diff --git a/tests/test_collectors_deep.py b/tests/test_collectors_deep.py
index d0b590e..90d3f92 100644
--- a/tests/test_collectors_deep.py
+++ b/tests/test_collectors_deep.py
@@ -146,54 +146,42 @@ def test_get_per_node_error(self):
 
 
 class TestTraffmonetizerDeep:
-    def test_collect_with_email_auth(self):
+    def test_collect_with_valid_token(self):
         from app.collectors.traffmonetizer import TraffmonetizerCollector
 
-        login_resp = _mock_response(200, {"data": {"token": "jwt-tok"}})
         balance_resp = _mock_response(200, {"data": {"balance": 1.50}})
 
         client = _make_async_client()
-        client.post.return_value = login_resp
         client.get.return_value = balance_resp
 
         with patch("app.collectors.traffmonetizer.httpx.AsyncClient", return_value=client):
-            c = TraffmonetizerCollector(email="test@test.com", password="pass")
+            c = TraffmonetizerCollector(token="valid-jwt")
             result = asyncio.run(c.collect())
         assert result.error is None
         assert result.balance == 1.50
 
-    def test_collect_token_refresh_on_401(self):
+    def test_collect_token_expired(self):
         from app.collectors.traffmonetizer import TraffmonetizerCollector
 
-        expired_resp = MagicMock()
-        expired_resp.status_code = 401
-        login_resp = _mock_response(200, {"data": {"token": "new-tok"}})
-        ok_resp = _mock_response(200, {"data": {"balance": 0.75}})
+        resp_401 = MagicMock()
+        resp_401.status_code = 401
 
         client = _make_async_client()
-        client.post.return_value = login_resp
-        client.get.side_effect = [expired_resp, ok_resp]
+        client.get.return_value = resp_401
 
         with patch("app.collectors.traffmonetizer.httpx.AsyncClient", return_value=client):
-            c = TraffmonetizerCollector(email="test@test.com", password="pass", token="old-tok")
+            c = TraffmonetizerCollector(token="expired-jwt")
             result = asyncio.run(c.collect())
-        assert result.error is None
-        assert result.balance == 0.75
+        assert result.error is not None
+        assert "expired" in result.error.lower()
 
-    def test_collect_auth_permanently_failed(self):
+    def test_collect_no_token(self):
         from app.collectors.traffmonetizer import TraffmonetizerCollector
 
-        resp_401 = MagicMock()
-        resp_401.status_code = 401
-
-        client = _make_async_client()
-        client.get.return_value = resp_401
-        # No email set, so no re-auth possible
-        with patch("app.collectors.traffmonetizer.httpx.AsyncClient", return_value=client):
-            c = TraffmonetizerCollector(token="expired-tok")
-            result = asyncio.run(c.collect())
+        c = TraffmonetizerCollector(token="")
+        result = asyncio.run(c.collect())
         assert result.error is not None
-        assert "Authentication" in result.error
+        assert "No token" in result.error
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py
index d64477d..a651540 100644
--- a/tests/test_coverage_gaps.py
+++ b/tests/test_coverage_gaps.py
@@ -914,18 +914,20 @@ def test_bitping_login_missing_cookie(self):
             result = asyncio.run(c.collect())
         assert result.error is not None
 
-    def test_traffmonetizer_login_missing_token(self):
-        """Cover traffmonetizer.py line 45: login response without token."""
+    def test_traffmonetizer_403_returns_expired_error(self):
+        """Cover traffmonetizer.py: 403 response returns token expired error."""
         from app.collectors.traffmonetizer import TraffmonetizerCollector
 
-        login_resp = _mock_response(200, {"data": {}})
+        resp_403 = MagicMock()
+        resp_403.status_code = 403
         client = _make_async_client()
-        client.post.return_value = login_resp
+        client.get.return_value = resp_403
 
         with patch("app.collectors.traffmonetizer.httpx.AsyncClient", return_value=client):
-            c = TraffmonetizerCollector(email="test@test.com", password="pass")
+            c = TraffmonetizerCollector(token="some-jwt")
             result = asyncio.run(c.collect())
         assert isinstance(result, EarningsResult)
+        assert "expired" in result.error.lower()
 
     def test_repocket_login_missing_token(self):
         """Cover repocket.py lines 49, 55: login without idToken."""

From 548bf080d5e4359bbe685193ef48af6ee5a384c4 Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 14:31:20 +0200
Subject: [PATCH 2/9] test: add orchestrator volume cleanup + delete_volumes
 route tests

---
 tests/test_coverage_gaps.py      | 110 +++++++++++++++++++++++++++++++
 tests/test_main_deploy_routes.py |  15 +++++
 2 files changed, 125 insertions(+)

diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py
index a651540..2ee71c2 100644
--- a/tests/test_coverage_gaps.py
+++ b/tests/test_coverage_gaps.py
@@ -1089,3 +1089,113 @@ def test_resolve_secret_key_oserror_reading(self, tmp_path):
             result = auth._resolve_secret_key()
             # Should generate a new key since reading failed
             assert len(result) > 20
+
+
+# ---------------------------------------------------------------------------
+# Orchestrator: volume cleanup on remove
+# ---------------------------------------------------------------------------
+
+
+class TestOrchestratorVolumeCleanup:
+    """Cover orchestrator.remove_service with delete_volumes flag."""
+
+    def test_remove_without_volumes(self):
+        from app import orchestrator
+
+        container = MagicMock()
+        container.name = "cashpilot-test"
+        container.attrs = {"Mounts": []}
+
+        with patch.object(orchestrator, "_find_container", return_value=container):
+            result = orchestrator.remove_service("test")
+
+        container.remove.assert_called_once_with(force=True)
+        assert result["container"] == "cashpilot-test"
+        assert result["deleted_volumes"] == []
+        assert result["failed_volumes"] == []
+
+    def test_remove_with_delete_volumes_named(self):
+        from app import orchestrator
+
+        container = MagicMock()
+        container.name = "cashpilot-honeygain"
+        container.attrs = {
+            "Mounts": [
+                {"Type": "volume", "Name": "honeygain_data"},
+                {"Type": "bind", "Source": "/host/path"},
+                {"Type": "volume", "Name": "honeygain_config"},
+            ]
+        }
+
+        mock_vol = MagicMock()
+        mock_client = MagicMock()
+        mock_client.volumes.get.return_value = mock_vol
+
+        with (
+            patch.object(orchestrator, "_find_container", return_value=container),
+            patch.object(orchestrator, "_get_client", return_value=mock_client),
+        ):
+            result = orchestrator.remove_service("honeygain", delete_volumes=True)
+
+        container.remove.assert_called_once_with(force=True)
+        assert "honeygain_data" in result["deleted_volumes"]
+        assert "honeygain_config" in result["deleted_volumes"]
+        assert result["failed_volumes"] == []
+        assert mock_vol.remove.call_count == 2
+
+    def test_remove_volume_not_found_counts_as_deleted(self):
+        from docker.errors import NotFound
+
+        from app import orchestrator
+
+        container = MagicMock()
+        container.name = "cashpilot-test"
+        container.attrs = {"Mounts": [{"Type": "volume", "Name": "gone_vol"}]}
+
+        mock_client = MagicMock()
+        mock_client.volumes.get.side_effect = NotFound("gone")
+
+        with (
+            patch.object(orchestrator, "_find_container", return_value=container),
+            patch.object(orchestrator, "_get_client", return_value=mock_client),
+        ):
+            result = orchestrator.remove_service("test", delete_volumes=True)
+
+        assert "gone_vol" in result["deleted_volumes"]
+        assert result["failed_volumes"] == []
+
+    def test_remove_volume_api_error_recorded(self):
+        from docker.errors import APIError
+
+        from app import orchestrator
+
+        container = MagicMock()
+        container.name = "cashpilot-test"
+        container.attrs = {"Mounts": [{"Type": "volume", "Name": "busy_vol"}]}
+
+        mock_vol = MagicMock()
+        mock_vol.remove.side_effect = APIError("volume in use")
+        mock_client = MagicMock()
+        mock_client.volumes.get.return_value = mock_vol
+
+        with (
+            patch.object(orchestrator, "_find_container", return_value=container),
+            patch.object(orchestrator, "_get_client", return_value=mock_client),
+        ):
+            result = orchestrator.remove_service("test", delete_volumes=True)
+
+        assert result["deleted_volumes"] == []
+        assert "busy_vol" in result["failed_volumes"]
+
+    def test_delete_volumes_false_skips_volume_inspection(self):
+        from app import orchestrator
+
+        container = MagicMock()
+        container.name = "cashpilot-test"
+        container.attrs = {"Mounts": [{"Type": "volume", "Name": "keep_me"}]}
+
+        with patch.object(orchestrator, "_find_container", return_value=container):
+            result = orchestrator.remove_service("test", delete_volumes=False)
+
+        assert result["deleted_volumes"] == []
+        assert result["failed_volumes"] == []
diff --git a/tests/test_main_deploy_routes.py b/tests/test_main_deploy_routes.py
index ede5a44..4d5af80 100644
--- a/tests/test_main_deploy_routes.py
+++ b/tests/test_main_deploy_routes.py
@@ -269,6 +269,21 @@ def test_old_remove_route(self, client):
             resp = client.delete("/api/remove/honeygain")
             assert resp.status_code == 200
 
+    def test_remove_with_delete_volumes(self, client):
+        worker, mock_client = self._setup_proxy()
+        with (
+            _auth_writer(),
+            patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=[worker]),
+            patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker),
+            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"),
+        ):
+            resp = client.delete("/api/services/honeygain?delete_volumes=true")
+            assert resp.status_code == 200
+            call_args = mock_client.delete.call_args
+            assert "delete_volumes" in str(call_args)
+
 
 # ---------------------------------------------------------------------------
 # Worker proxy error handling

From 9fad93879a52c610f33310ba79bf0b1378b4e899 Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 14:45:36 +0200
Subject: [PATCH 3/9] fix: security hardening and reliability improvements

- Use hmac.compare_digest for admin/fleet API key comparison (timing attack)
- Add Content-Security-Policy header to security middleware
- Add login rate limiting (5 attempts per 5 minutes per IP)
- Block privileged containers and dangerous volume mounts on worker
- Escape error messages in fleet.html to prevent XSS
- Add PRAGMA busy_timeout=5000 to prevent SQLITE_BUSY on contention
- Parallelize collector execution with asyncio.gather
- Add collection lock to prevent overlapping runs
- Replace silent contextlib.suppress with try/except+logging on worker
- Upgrade background task error logging from debug to warning
---
 app/auth.py              |  7 +++-
 app/database.py          |  1 +
 app/main.py              | 90 ++++++++++++++++++++++++++++------------
 app/templates/fleet.html |  3 +-
 app/worker_api.py        | 29 +++++++++++--
 5 files changed, 97 insertions(+), 33 deletions(-)

diff --git a/app/auth.py b/app/auth.py
index 9794ffc..4b25eb3 100644
--- a/app/auth.py
+++ b/app/auth.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import hmac
 import logging
 import os
 import secrets
@@ -108,10 +109,12 @@ def get_current_user(request: Request) -> dict[str, Any] | None:
     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}":
+        if admin_key and hmac.compare_digest(auth_header.encode(), f"Bearer {admin_key}".encode()):
             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}":
+        if resolved_fleet_key and hmac.compare_digest(
+            auth_header.encode(), f"Bearer {resolved_fleet_key}".encode()
+        ):
             return {"uid": 0, "u": "fleet", "r": "fleet"}
 
     # Fall back to session cookie
diff --git a/app/database.py b/app/database.py
index 56a9586..cc50a35 100644
--- a/app/database.py
+++ b/app/database.py
@@ -172,6 +172,7 @@ async def _get_db() -> aiosqlite.Connection:
     db.row_factory = aiosqlite.Row
     await db.execute("PRAGMA journal_mode=WAL")
     await db.execute("PRAGMA foreign_keys=ON")
+    await db.execute("PRAGMA busy_timeout=5000")
     return db
 
 
diff --git a/app/main.py b/app/main.py
index 8f68592..37005be 100644
--- a/app/main.py
+++ b/app/main.py
@@ -14,8 +14,10 @@
 import logging
 import os
 import re
+from collections import defaultdict
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime, timedelta
+from time import monotonic
 from typing import Any
 from urllib.parse import urlparse
 
@@ -40,6 +42,21 @@
 
 # In-memory store for the latest collector alerts (errors from last run)
 _collector_alerts: list[dict[str, str]] = []
+_collection_lock = asyncio.Lock()
+
+# Login rate limiting
+_login_attempts: dict[str, list[float]] = defaultdict(list)
+_LOGIN_MAX_ATTEMPTS = 5
+_LOGIN_WINDOW_SECONDS = 300
+
+
+def _check_login_rate(client_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:
+        raise HTTPException(status_code=429, detail="Too many login attempts. Try again in a few minutes.")
+    _login_attempts[client_ip].append(now)
 
 
 def _safe_json(raw: str, fallback: Any = None) -> Any:
@@ -155,36 +172,45 @@ async def _run_health_check() -> None:
             else:
                 await database.record_health_event(slug, "check_down", status)
     except Exception as exc:
-        logger.debug("Health check skipped: %s", exc)
+        logger.warning("Health check skipped: %s", exc)
 
 
 async def _run_collection() -> None:
     """Collect earnings from all deployed services that have collectors."""
     global _collector_alerts
-    try:
-        deployments = await database.get_deployments()
-        config = await database.get_config() or {}
-        if not isinstance(config, dict):
-            config = {}
-        from app.collectors import make_collectors
-
-        collectors = make_collectors(deployments, config)
-        alerts: list[dict[str, str]] = []
-        for collector in collectors:
-            result = await collector.collect()
-            if result.error:
-                logger.warning("Collection error for %s: %s", result.platform, result.error)
-                alerts.append({"platform": result.platform, "error": result.error})
-            else:
-                await database.upsert_earnings(
-                    platform=result.platform,
-                    balance=result.balance,
-                    currency=result.currency,
-                )
-                logger.info("Collected %s: %.4f %s", result.platform, result.balance, result.currency)
-        _collector_alerts = alerts
-    except Exception as exc:
-        logger.error("Collection run failed: %s", exc)
+    if _collection_lock.locked():
+        logger.info("Collection already in progress, skipping")
+        return
+    async with _collection_lock:
+        try:
+            deployments = await database.get_deployments()
+            config = await database.get_config() or {}
+            if not isinstance(config, dict):
+                config = {}
+            from app.collectors import make_collectors
+
+            collectors = make_collectors(deployments, config)
+            results = await asyncio.gather(
+                *(c.collect() for c in collectors), return_exceptions=True
+            )
+            alerts: list[dict[str, str]] = []
+            for result in results:
+                if isinstance(result, Exception):
+                    logger.warning("Collector raised exception: %s", result)
+                    continue
+                if result.error:
+                    logger.warning("Collection error for %s: %s", result.platform, result.error)
+                    alerts.append({"platform": result.platform, "error": result.error})
+                else:
+                    await database.upsert_earnings(
+                        platform=result.platform,
+                        balance=result.balance,
+                        currency=result.currency,
+                    )
+                    logger.info("Collected %s: %.4f %s", result.platform, result.balance, result.currency)
+            _collector_alerts = alerts
+        except Exception as exc:
+            logger.error("Collection run failed: %s", exc)
 
 
 # ---------------------------------------------------------------------------
@@ -199,7 +225,7 @@ async def _run_data_retention() -> None:
         if deleted:
             logger.info("Data retention: purged %d old rows", deleted)
     except Exception as exc:
-        logger.debug("Data retention error: %s", exc)
+        logger.warning("Data retention error: %s", exc)
 
 
 async def _check_stale_workers() -> None:
@@ -214,7 +240,7 @@ async def _check_stale_workers() -> None:
                     await database.set_worker_status(w["id"], "offline")
                     logger.info("Worker '%s' marked offline (last heartbeat: %s)", w["name"], w["last_heartbeat"])
     except Exception as exc:
-        logger.debug("Stale worker check error: %s", exc)
+        logger.warning("Stale worker check error: %s", exc)
 
 
 FLEET_API_KEY = fleet_key.resolve_fleet_key()
@@ -261,6 +287,15 @@ async def dispatch(self, request, call_next):
         response.headers["X-Frame-Options"] = "DENY"
         response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
         response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
+        response.headers["Content-Security-Policy"] = (
+            "default-src 'self'; "
+            "script-src 'self' https://cdn.jsdelivr.net 'unsafe-inline'; "
+            "style-src 'self' https://fonts.googleapis.com 'unsafe-inline'; "
+            "font-src 'self' https://fonts.gstatic.com; "
+            "img-src 'self' data:; "
+            "connect-src 'self'; "
+            "frame-ancestors 'none'"
+        )
         return response
 
 
@@ -356,6 +391,7 @@ async def do_login(
     username: str = Form(...),
     password: str = Form(...),
 ):
+    _check_login_rate(request.client.host if request.client else "unknown")
     user = await database.get_user_by_username(username)
     if not user or not auth.verify_password(password, user["password"]):
         return templates.TemplateResponse(
diff --git a/app/templates/fleet.html b/app/templates/fleet.html
index 31203b7..d2f8b60 100644
--- a/app/templates/fleet.html
+++ b/app/templates/fleet.html
@@ -89,8 +89,9 @@ <h2 class="section-title">Workers</h2>
 
       renderWorkers(workers);
     } catch (err) {
+      const msg = (err.message || '').replace(/[&<>"']/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'})[c]);
       document.getElementById('fleet-workers').innerHTML =
-        `<div class="empty-state"><div class="empty-state-text">Failed to load fleet data: ${err.message}</div></div>`;
+        `<div class="empty-state"><div class="empty-state-text">Failed to load fleet data: ${msg}</div></div>`;
     }
   }
 
diff --git a/app/worker_api.py b/app/worker_api.py
index bd00877..3a1753e 100644
--- a/app/worker_api.py
+++ b/app/worker_api.py
@@ -80,8 +80,10 @@ async def _send_heartbeat() -> None:
     global _ui_connected, _last_heartbeat, _last_error
 
     containers = []
-    with contextlib.suppress(Exception):
+    try:
         containers = orchestrator.get_status()
+    except Exception as exc:
+        logger.warning("Failed to get container status for heartbeat: %s", exc)
 
     payload = {
         "name": WORKER_NAME,
@@ -177,8 +179,10 @@ async def lifespan(app: FastAPI):
 async def worker_status_page():
     """Self-contained HTML status page for the worker."""
     containers = []
-    with contextlib.suppress(Exception):
+    try:
         containers = orchestrator.get_status_cached()
+    except Exception as exc:
+        logger.warning("Failed to get container status for status page: %s", exc)
 
     container_rows = ""
     for c in containers:
@@ -267,13 +271,31 @@ class DeploySpec(BaseModel):
     labels: dict[str, str] = {}
 
 
+_BLOCKED_VOLUME_SOURCES = {"/", "/etc", "/var/run/docker.sock", "/root", "/proc", "/sys"}
+_BLOCKED_CAPS = {"ALL", "SYS_ADMIN", "SYS_PTRACE"}
+
+
+def _validate_deploy_spec(spec: DeploySpec) -> None:
+    if spec.privileged:
+        raise HTTPException(status_code=403, detail="Privileged containers are not allowed")
+    if spec.cap_add:
+        blocked = _BLOCKED_CAPS & {c.upper() for c in spec.cap_add}
+        if blocked:
+            raise HTTPException(status_code=403, detail=f"Blocked capabilities: {', '.join(blocked)}")
+    for source in spec.volumes:
+        if source.rstrip("/") in _BLOCKED_VOLUME_SOURCES or source == "/":
+            raise HTTPException(status_code=403, detail=f"Volume mount '{source}' is blocked")
+
+
 @app.get("/api/status")
 async def api_worker_status(request: Request) -> dict[str, Any]:
     """Return worker status summary."""
     _verify_api_key(request)
     containers = []
-    with contextlib.suppress(Exception):
+    try:
         containers = orchestrator.get_status_cached()
+    except Exception as exc:
+        logger.warning("Failed to get container status: %s", exc)
     return {
         "name": WORKER_NAME,
         "docker_available": orchestrator.docker_available(),
@@ -297,6 +319,7 @@ async def api_list_containers(request: Request) -> list[dict[str, Any]]:
 async def api_deploy_container(request: Request, slug: str, spec: DeploySpec) -> dict[str, str]:
     """Deploy a container from spec sent by UI."""
     _verify_api_key(request)
+    _validate_deploy_spec(spec)
     try:
         container_id = orchestrator.deploy_raw(
             slug=slug,

From ebc2c7ae052131df55e492f47d518f617acdee45 Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 14:51:57 +0200
Subject: [PATCH 4/9] fix: skip orchestrator tests when docker SDK unavailable

CI environment doesn't install docker SDK (runtime-only dependency).
Use pytest.importorskip to gracefully skip these tests.
---
 tests/test_coverage_gaps.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py
index 2ee71c2..1d204d6 100644
--- a/tests/test_coverage_gaps.py
+++ b/tests/test_coverage_gaps.py
@@ -10,10 +10,11 @@
 from contextlib import asynccontextmanager
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import pytest
+
 os.environ.setdefault("CASHPILOT_API_KEY", "test-fleet-key")
 
 import httpx
-import pytest
 import yaml
 
 from app import catalog, database
@@ -1096,6 +1097,9 @@ def test_resolve_secret_key_oserror_reading(self, tmp_path):
 # ---------------------------------------------------------------------------
 
 
+docker = pytest.importorskip("docker")
+
+
 class TestOrchestratorVolumeCleanup:
     """Cover orchestrator.remove_service with delete_volumes flag."""
 

From d75e702b4cbf67555f87d5d7e3998958900e82bd Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 14:53:40 +0200
Subject: [PATCH 5/9] style: apply ruff format to modified files

---
 app/auth.py       | 4 +---
 app/main.py       | 8 ++------
 app/worker_api.py | 4 +---
 3 files changed, 4 insertions(+), 12 deletions(-)

diff --git a/app/auth.py b/app/auth.py
index 4b25eb3..0e4b0d3 100644
--- a/app/auth.py
+++ b/app/auth.py
@@ -112,9 +112,7 @@ def get_current_user(request: Request) -> dict[str, Any] | None:
         if admin_key and hmac.compare_digest(auth_header.encode(), f"Bearer {admin_key}".encode()):
             return {"uid": 0, "u": "api", "r": "owner"}
         resolved_fleet_key = _fleet_key_mod.resolve_fleet_key()
-        if resolved_fleet_key and hmac.compare_digest(
-            auth_header.encode(), f"Bearer {resolved_fleet_key}".encode()
-        ):
+        if resolved_fleet_key and hmac.compare_digest(auth_header.encode(), f"Bearer {resolved_fleet_key}".encode()):
             return {"uid": 0, "u": "fleet", "r": "fleet"}
 
     # Fall back to session cookie
diff --git a/app/main.py b/app/main.py
index 37005be..08a3402 100644
--- a/app/main.py
+++ b/app/main.py
@@ -190,9 +190,7 @@ async def _run_collection() -> None:
             from app.collectors import make_collectors
 
             collectors = make_collectors(deployments, config)
-            results = await asyncio.gather(
-                *(c.collect() for c in collectors), return_exceptions=True
-            )
+            results = await asyncio.gather(*(c.collect() for c in collectors), return_exceptions=True)
             alerts: list[dict[str, str]] = []
             for result in results:
                 if isinstance(result, Exception):
@@ -983,9 +981,7 @@ async def _proxy_worker_command(
     try:
         async with httpx.AsyncClient(timeout=30) as client:
             if command == "remove":
-                resp = await client.delete(
-                    f"{url}/api/containers/{slug}", headers=headers, params=params
-                )
+                resp = await client.delete(f"{url}/api/containers/{slug}", headers=headers, params=params)
             else:
                 resp = await client.post(f"{url}/api/containers/{slug}/{command}", headers=headers)
             if resp.status_code >= 400:
diff --git a/app/worker_api.py b/app/worker_api.py
index 3a1753e..9386963 100644
--- a/app/worker_api.py
+++ b/app/worker_api.py
@@ -377,9 +377,7 @@ async def api_start_container(request: Request, slug: str) -> dict[str, str]:
 
 
 @app.delete("/api/containers/{slug}")
-async def api_remove_container(
-    request: Request, slug: str, delete_volumes: bool = False
-) -> dict[str, Any]:
+async def api_remove_container(request: Request, slug: str, delete_volumes: bool = False) -> dict[str, Any]:
     _verify_api_key(request)
     try:
         result = orchestrator.remove_service(slug, delete_volumes=delete_volumes)

From 68df22558603f15ae58e1a15edf5aa2360d6743b Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 15:11:13 +0200
Subject: [PATCH 6/9] =?UTF-8?q?feat:=20collector=20base=20refactor=20?=
 =?UTF-8?q?=E2=80=94=20reusable=20clients,=20retry,=20caching,=20async=20o?=
 =?UTF-8?q?rchestrator?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- BaseCollector: add _get_client() for HTTP connection reuse, _retry()
  for transient failure recovery, close() for lifecycle management
- Factory: cache collector instances across cycles (tokens persist),
  evict stale instances on config change, close_all on shutdown
- All 14 collectors: use shared client, add exc_info=True to error logs,
  add super().__init__(), use self._retry() for balance fetches
- Grass: replace magic sentinel floats with _GrassStatus enum
- IPRoyal: stable device_id across logins, token caching with 401 retry
- EarnApp: configurable API version via class attribute
- Bytelixir: path-based URL validation instead of string "login" check
- Orchestrator: wrap all Docker SDK calls in asyncio.to_thread()
- Auth: add session iat (issued-at) + SESSION_EPOCH for mass invalidation
- Database: add password_changed_at column migration + update function
- Pipeline: close stale collectors, sanitize alert errors (200 char max)
- Password policy: minimum raised from 8 to 10 characters
- Remove dead bytes_uploaded field from EarningsResult
---
 app/auth.py                       |  13 ++-
 app/collectors/__init__.py        |  59 +++++++++---
 app/collectors/base.py            |  42 ++++++++-
 app/collectors/bitping.py         |  44 ++++-----
 app/collectors/bytelixir.py       | 144 ++++++++++++++++--------------
 app/collectors/earnapp.py         |  85 ++++++++++--------
 app/collectors/earnfm.py          |  44 ++++-----
 app/collectors/grass.py           |  73 ++++++++-------
 app/collectors/honeygain.py       |  44 ++++-----
 app/collectors/iproyal.py         |  59 +++++++-----
 app/collectors/mystnodes.py       |  93 +++++++++----------
 app/collectors/packetstream.py    | 101 +++++++++++----------
 app/collectors/proxyrack.py       |  10 ++-
 app/collectors/repocket.py        |  54 +++++------
 app/collectors/salad.py           |  37 ++++----
 app/collectors/storj.py           |  63 ++++++-------
 app/collectors/traffmonetizer.py  |  41 +++++----
 app/database.py                   |  22 +++++
 app/main.py                       |  24 +++--
 app/worker_api.py                 |  21 ++---
 tests/test_collectors.py          |   1 -
 tests/test_collectors_deep.py     |  10 ++-
 tests/test_collectors_extended.py |  12 ++-
 tests/test_coverage_gaps.py       |  14 +--
 24 files changed, 658 insertions(+), 452 deletions(-)

diff --git a/app/auth.py b/app/auth.py
index 0e4b0d3..7665b20 100644
--- a/app/auth.py
+++ b/app/auth.py
@@ -9,6 +9,7 @@
 import logging
 import os
 import secrets
+import time
 from pathlib import Path
 from typing import Any
 
@@ -76,6 +77,10 @@ def _resolve_secret_key() -> str:
 
 _serializer = URLSafeTimedSerializer(SECRET_KEY)
 
+# Global session epoch: tokens issued before this timestamp are rejected.
+# Bump via env var to mass-invalidate all sessions (e.g. after a credential leak).
+_SESSION_EPOCH = float(os.getenv("CASHPILOT_SESSION_EPOCH", "0"))
+
 
 def hash_password(password: str) -> str:
     # bcrypt enforces a 72-byte limit; truncate UTF-8 bytes (not characters)
@@ -89,12 +94,16 @@ def verify_password(password: str, hashed: str) -> bool:
 
 
 def create_session_token(user_id: int, username: str, role: str) -> str:
-    return _serializer.dumps({"uid": user_id, "u": username, "r": role})
+    return _serializer.dumps({"uid": user_id, "u": username, "r": role, "iat": time.time()})
 
 
 def decode_session_token(token: str) -> dict[str, Any] | None:
     try:
-        return _serializer.loads(token, max_age=SESSION_MAX_AGE)
+        data = _serializer.loads(token, max_age=SESSION_MAX_AGE)
+        # Reject tokens issued before session epoch (for mass invalidation)
+        if data.get("iat", 0) < _SESSION_EPOCH:
+            return None
+        return data
     except (BadSignature, Exception):
         return None
 
diff --git a/app/collectors/__init__.py b/app/collectors/__init__.py
index 438029e..1f470fd 100644
--- a/app/collectors/__init__.py
+++ b/app/collectors/__init__.py
@@ -63,22 +63,30 @@
     "salad": ["auth_cookie"],
 }
 
+_cached_collectors: dict[str, BaseCollector] = {}
+_cached_kwargs: dict[str, dict[str, str]] = {}
+_stale: list[BaseCollector] = []
+
+
+async def _close_stale() -> None:
+    """Close collectors evicted from cache due to config changes."""
+    global _stale
+    for c in _stale:
+        await c.close()
+    _stale = []
+
 
 def make_collectors(
     deployments: list[dict[str, Any]],
     config: dict[str, str],
 ) -> list[BaseCollector]:
-    """Create collector instances for all deployed services that have collectors.
+    """Create or retrieve cached collector instances for deployed services.
 
-    Args:
-        deployments: List of deployment dicts (must have 'slug' key).
-        config: User config dict. Collector credentials are stored as
-                ``{slug}_email``, ``{slug}_password``, etc.
-
-    Returns:
-        List of ready-to-use collector instances.
+    Reuses a cached instance when the resolved kwargs for a slug match
+    the previous invocation. Evicts stale instances when config changes.
     """
     collectors: list[BaseCollector] = []
+    active_slugs: set[str] = set()
 
     for dep in deployments:
         slug = dep.get("slug", "")
@@ -89,7 +97,6 @@ def make_collectors(
         arg_keys = _COLLECTOR_ARGS.get(slug, [])
 
         # Resolve constructor kwargs from config
-        # Args prefixed with ? are optional
         kwargs: dict[str, str] = {}
         missing: list[str] = []
         for arg in arg_keys:
@@ -110,18 +117,50 @@ def make_collectors(
             )
             continue
 
+        active_slugs.add(slug)
+
+        # Reuse cached instance if kwargs unchanged
+        if slug in _cached_collectors and _cached_kwargs.get(slug) == kwargs:
+            collectors.append(_cached_collectors[slug])
+            logger.debug("Reusing cached collector for %s", slug)
+            continue
+
+        # Config changed or new slug — evict old instance
+        if slug in _cached_collectors:
+            _stale.append(_cached_collectors[slug])
+
         try:
-            collectors.append(cls(**kwargs))
+            instance = cls(**kwargs)
+            _cached_collectors[slug] = instance
+            _cached_kwargs[slug] = kwargs
+            collectors.append(instance)
             logger.debug("Created collector for %s", slug)
         except Exception as exc:
             logger.error("Failed to create collector for %s: %s", slug, exc)
 
+    # Evict collectors for slugs no longer deployed
+    for slug in list(_cached_collectors.keys()):
+        if slug not in active_slugs:
+            _stale.append(_cached_collectors.pop(slug))
+            _cached_kwargs.pop(slug, None)
+
     return collectors
 
 
+async def close_all_collectors() -> None:
+    """Close all cached collector HTTP clients and clear the cache."""
+    global _cached_collectors, _cached_kwargs
+    for collector in _cached_collectors.values():
+        await collector.close()
+    _cached_collectors = {}
+    _cached_kwargs = {}
+    await _close_stale()
+
+
 __all__ = [
     "BaseCollector",
     "EarningsResult",
     "COLLECTOR_MAP",
     "make_collectors",
+    "close_all_collectors",
 ]
diff --git a/app/collectors/base.py b/app/collectors/base.py
index 12b8a6c..b2a0609 100644
--- a/app/collectors/base.py
+++ b/app/collectors/base.py
@@ -2,7 +2,14 @@
 
 from __future__ import annotations
 
+import asyncio
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
+from typing import Any, TypeVar
+
+import httpx
+
+T = TypeVar("T")
 
 
 @dataclass
@@ -12,7 +19,6 @@ class EarningsResult:
     platform: str
     balance: float
     currency: str = "USD"
-    bytes_uploaded: int = 0
     error: str | None = None
 
 
@@ -24,5 +30,39 @@ class BaseCollector:
 
     platform: str = ""
 
+    def __init__(self) -> None:
+        self._client: httpx.AsyncClient | None = None
+
+    def _get_client(self, **kwargs: Any) -> httpx.AsyncClient:
+        """Return a reusable httpx client, creating one if needed."""
+        if self._client is None or self._client.is_closed:
+            defaults: dict[str, Any] = {"timeout": 30}
+            defaults.update(kwargs)
+            self._client = httpx.AsyncClient(**defaults)
+        return self._client
+
+    async def close(self) -> None:
+        """Close the underlying HTTP client. Safe to call multiple times."""
+        if self._client is not None and not self._client.is_closed:
+            await self._client.aclose()
+            self._client = None
+
+    async def _retry(
+        self,
+        coro_fn: Callable[[], Awaitable[T]],
+        max_retries: int = 2,
+        backoff: float = 1.0,
+    ) -> T:
+        """Retry a coroutine on transient network failures."""
+        last_exc: BaseException | None = None
+        for attempt in range(max_retries + 1):
+            try:
+                return await coro_fn()
+            except (httpx.TimeoutException, httpx.NetworkError) as exc:
+                last_exc = exc
+                if attempt < max_retries:
+                    await asyncio.sleep(backoff * (2**attempt))
+        raise last_exc  # type: ignore[misc]
+
     async def collect(self) -> EarningsResult:
         raise NotImplementedError
diff --git a/app/collectors/bitping.py b/app/collectors/bitping.py
index adaa397..97acfc9 100644
--- a/app/collectors/bitping.py
+++ b/app/collectors/bitping.py
@@ -23,6 +23,7 @@ class BitpingCollector(BaseCollector):
     platform = "bitping"
 
     def __init__(self, email: str, password: str) -> None:
+        super().__init__()
         self.email = email
         self.password = password
         self._token: str | None = None
@@ -51,36 +52,37 @@ async def _authenticate(self, client: httpx.AsyncClient) -> str:
     async def collect(self) -> EarningsResult:
         """Fetch current Bitping earnings."""
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
-                if not self._token:
-                    self._token = await self._authenticate(client)
+            client = self._get_client(timeout=30)
+            if not self._token:
+                self._token = await self._authenticate(client)
 
-                headers = {"Authorization": f"Bearer {self._token}"}
-                resp = await client.get(
+            headers = {"Authorization": f"Bearer {self._token}"}
+
+            async def _fetch_balance() -> httpx.Response:
+                return await client.get(
                     f"{API_BASE}/api/v2/payouts/earnings",
                     headers=headers,
                 )
 
-                if resp.status_code == 401:
-                    self._token = await self._authenticate(client)
-                    headers = {"Authorization": f"Bearer {self._token}"}
-                    resp = await client.get(
-                        f"{API_BASE}/api/v2/payouts/earnings",
-                        headers=headers,
-                    )
+            resp = await self._retry(_fetch_balance)
 
-                resp.raise_for_status()
-                data = resp.json()
+            if resp.status_code == 401:
+                self._token = await self._authenticate(client)
+                headers = {"Authorization": f"Bearer {self._token}"}
+                resp = await self._retry(_fetch_balance)
 
-                balance = float(data.get("usdEarnings", 0))
+            resp.raise_for_status()
+            data = resp.json()
 
-                return EarningsResult(
-                    platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
-                )
+            balance = float(data.get("usdEarnings", 0))
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("Bitping collection failed: %s", exc)
+            logger.error("Bitping collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/bytelixir.py b/app/collectors/bytelixir.py
index 232146e..1e71438 100644
--- a/app/collectors/bytelixir.py
+++ b/app/collectors/bytelixir.py
@@ -21,6 +21,7 @@
 
 import logging
 import re
+from typing import Any
 from urllib.parse import unquote
 
 import httpx
@@ -46,6 +47,7 @@ def __init__(
         remember_web: str = "",
         xsrf_token: str = "",
     ) -> None:
+        super().__init__()
         self.session_cookie = unquote(session_cookie)
         self.remember_web = unquote(remember_web) if remember_web else ""
         self.xsrf_token = unquote(xsrf_token) if xsrf_token else ""
@@ -54,31 +56,33 @@ def __init__(
     # Internal helpers
     # ------------------------------------------------------------------
 
-    def _make_client(self) -> httpx.AsyncClient:
-        """Build an httpx client with all session cookies pre-set."""
-        cookies = httpx.Cookies()
-        cookies.set(
-            "bytelixir_session",
-            self.session_cookie,
-            domain="dash.bytelixir.com",
-        )
-        if self.remember_web:
+    def _get_client(self, **kwargs: Any) -> httpx.AsyncClient:
+        """Return a reusable httpx client with all session cookies pre-set."""
+        if self._client is None or self._client.is_closed:
+            cookies = httpx.Cookies()
             cookies.set(
-                self._REMEMBER_COOKIE,
-                self.remember_web,
+                "bytelixir_session",
+                self.session_cookie,
                 domain="dash.bytelixir.com",
             )
-        if self.xsrf_token:
-            cookies.set(
-                "XSRF-TOKEN",
-                self.xsrf_token,
-                domain="dash.bytelixir.com",
+            if self.remember_web:
+                cookies.set(
+                    self._REMEMBER_COOKIE,
+                    self.remember_web,
+                    domain="dash.bytelixir.com",
+                )
+            if self.xsrf_token:
+                cookies.set(
+                    "XSRF-TOKEN",
+                    self.xsrf_token,
+                    domain="dash.bytelixir.com",
+                )
+            self._client = httpx.AsyncClient(
+                timeout=30,
+                cookies=cookies,
+                follow_redirects=True,
             )
-        return httpx.AsyncClient(
-            timeout=30,
-            cookies=cookies,
-            follow_redirects=True,
-        )
+        return self._client
 
     @staticmethod
     def _browser_headers(*, accept: str = "text/html") -> dict[str, str]:
@@ -135,39 +139,45 @@ async def collect(self) -> EarningsResult:
         Fallback: ``/api/v1/user`` JSON endpoint.
         """
         try:
-            async with self._make_client() as client:
-                # --- 1. Try scraping the dashboard HTML ---------------
-                # Use the root URL — when authenticated it redirects to
-                # the localised dashboard (e.g. /en).  When not
-                # authenticated it redirects to /login.
-                html_resp = await client.get(
+            client = self._get_client()
+
+            # --- 1. Try scraping the dashboard HTML ---------------
+            # Use the root URL — when authenticated it redirects to
+            # the localised dashboard (e.g. /en).  When not
+            # authenticated it redirects to /login.
+            async def _fetch_html() -> httpx.Response:
+                return await client.get(
                     f"{API_BASE}/",
                     headers=self._browser_headers(),
                 )
 
-                # A 302 to /login means the session cookie is expired.
-                if "login" in str(html_resp.url):
+            html_resp = await self._retry(_fetch_html)
+
+            # A redirect away from a dashboard path means session expired.
+            final_path = html_resp.url.path
+            if not final_path.startswith(("/en", "/es", "/dashboard", "/home")):
+                return EarningsResult(
+                    platform=self.platform,
+                    balance=0.0,
+                    error="Session expired — refresh bytelixir_session cookie in Settings",
+                )
+
+            if html_resp.status_code == 200:
+                scraped = self._parse_balance_from_html(html_resp.text)
+                if scraped is not None:
+                    logger.info(
+                        "Bytelixir balance from HTML scrape: %s",
+                        scraped,
+                    )
                     return EarningsResult(
                         platform=self.platform,
-                        balance=0.0,
-                        error=("Session expired — refresh bytelixir_session cookie in Settings"),
+                        balance=round(scraped, 4),
+                        currency="USD",
                     )
 
-                if html_resp.status_code == 200:
-                    scraped = self._parse_balance_from_html(html_resp.text)
-                    if scraped is not None:
-                        logger.info(
-                            "Bytelixir balance from HTML scrape: %s",
-                            scraped,
-                        )
-                        return EarningsResult(
-                            platform=self.platform,
-                            balance=round(scraped, 4),
-                            currency="USD",
-                        )
-
-                # --- 2. Fallback: JSON API ----------------------------
-                api_resp = await client.get(
+            # --- 2. Fallback: JSON API ----------------------------
+            async def _fetch_api() -> httpx.Response:
+                return await client.get(
                     f"{API_BASE}/api/v1/user",
                     headers=self._browser_headers(accept="application/json")
                     | {
@@ -177,31 +187,33 @@ async def collect(self) -> EarningsResult:
                     },
                 )
 
-                if api_resp.status_code == 401:
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error=("Session expired — refresh bytelixir_session cookie in Settings"),
-                    )
-
-                api_resp.raise_for_status()
-                data = api_resp.json()
-
-                # Response shape: {"data": {"balance": "0.0000000000", ...}}
-                # NOTE: /api/v1/user returns *withdrawable* balance only, not
-                # total earned.  Flag this so the user knows it's approximate.
-                user_data = data.get("data", {})
-                balance_str = user_data.get("balance", "0")
-                balance = float(balance_str)
+            api_resp = await self._retry(_fetch_api)
 
+            if api_resp.status_code == 401:
                 return EarningsResult(
                     platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
-                    error="Withdrawable balance only (HTML scrape failed, using API fallback)",
+                    balance=0.0,
+                    error="Session expired — refresh bytelixir_session cookie in Settings",
                 )
+
+            api_resp.raise_for_status()
+            data = api_resp.json()
+
+            # Response shape: {"data": {"balance": "0.0000000000", ...}}
+            # NOTE: /api/v1/user returns *withdrawable* balance only, not
+            # total earned.  Flag this so the user knows it's approximate.
+            user_data = data.get("data", {})
+            balance_str = user_data.get("balance", "0")
+            balance = float(balance_str)
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+                error="Withdrawable balance only (HTML scrape failed, using API fallback)",
+            )
         except Exception as exc:
-            logger.error("Bytelixir collection failed: %s", exc)
+            logger.error("Bytelixir collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/earnapp.py b/app/collectors/earnapp.py
index fdd27ac..e323bf5 100644
--- a/app/collectors/earnapp.py
+++ b/app/collectors/earnapp.py
@@ -8,22 +8,23 @@
 
 import logging
 
-import httpx
+import httpx  # noqa: F401 — needed for test patching
 
 from app.collectors.base import BaseCollector, EarningsResult
 
 logger = logging.getLogger(__name__)
 
 API_BASE = "https://earnapp.com/dashboard/api"
-API_PARAMS = {"appid": "earnapp", "version": "1.613.719"}
 
 
 class EarnAppCollector(BaseCollector):
     """Collect earnings from EarnApp's dashboard API."""
 
     platform = "earnapp"
+    _API_VERSION = "1.613.719"
 
     def __init__(self, oauth_token: str, brd_sess_id: str = "") -> None:
+        super().__init__()
         self.oauth_token = oauth_token
         self.brd_sess_id = brd_sess_id
 
@@ -38,57 +39,67 @@ async def collect(self) -> EarningsResult:
             if self.brd_sess_id:
                 cookies["brd_sess_id"] = self.brd_sess_id
 
-            async with httpx.AsyncClient(timeout=30, cookies=cookies) as client:
-                # Step 1: Rotate XSRF token
-                await client.get(
-                    f"{API_BASE}/sec/rotate_xsrf",
-                    params=API_PARAMS,
-                )
-                xsrf_token = ""
-                for cookie_name, cookie_value in client.cookies.items():
-                    if cookie_name == "xsrf-token":
-                        xsrf_token = cookie_value
-                        break
-
-                # Step 2: Fetch balance
-                headers = {
-                    "X-Requested-With": "XMLHttpRequest",
-                }
-                if xsrf_token:
-                    headers["xsrf-token"] = xsrf_token
+            api_params = {"appid": "earnapp", "version": self._API_VERSION}
+            client = self._get_client(timeout=30, cookies=cookies)
+
+            # Step 1: Rotate XSRF token
+            await client.get(
+                f"{API_BASE}/sec/rotate_xsrf",
+                params=api_params,
+            )
+            xsrf_token = ""
+            for cookie_name, cookie_value in client.cookies.items():
+                if cookie_name == "xsrf-token":
+                    xsrf_token = cookie_value
+                    break
+
+            # Step 2: Fetch balance
+            headers = {
+                "X-Requested-With": "XMLHttpRequest",
+            }
+            if xsrf_token:
+                headers["xsrf-token"] = xsrf_token
 
+            async def _fetch_balance():
                 resp = await client.get(
                     f"{API_BASE}/money",
                     headers=headers,
-                    params=API_PARAMS,
+                    params=api_params,
                 )
 
                 if resp.status_code == 403:
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="Authentication failed — check OAuth token and session cookie",
-                    )
+                    return resp
 
                 resp.raise_for_status()
-                data = resp.json()
+                return resp
 
-                if "error" in data:
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error=data["error"],
-                    )
+            resp = await self._retry(_fetch_balance)
 
-                balance = float(data.get("balance", 0))
+            if resp.status_code == 403:
+                return EarningsResult(
+                    platform=self.platform,
+                    balance=0.0,
+                    error="Authentication failed — check OAuth token and session cookie",
+                )
 
+            data = resp.json()
+
+            if "error" in data:
                 return EarningsResult(
                     platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
+                    balance=0.0,
+                    error=data["error"],
                 )
+
+            balance = float(data.get("balance", 0))
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("EarnApp collection failed: %s", exc)
+            logger.error("EarnApp collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/earnfm.py b/app/collectors/earnfm.py
index 11d3a91..95204a8 100644
--- a/app/collectors/earnfm.py
+++ b/app/collectors/earnfm.py
@@ -30,6 +30,7 @@ class EarnFMCollector(BaseCollector):
     platform = "earnfm"
 
     def __init__(self, email: str, password: str) -> None:
+        super().__init__()
         self.email = email
         self.password = password
         self._access_token: str | None = None
@@ -76,36 +77,37 @@ async def _refresh(self, client: httpx.AsyncClient) -> str:
     async def collect(self) -> EarningsResult:
         """Fetch current Earn.fm balance."""
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
-                if not self._access_token:
-                    await self._authenticate(client)
+            client = self._get_client(timeout=30)
+            if not self._access_token:
+                await self._authenticate(client)
 
-                headers = {"X-API-Key": self._access_token}
-                resp = await client.get(
+            headers = {"X-API-Key": self._access_token}
+
+            async def _fetch_balance() -> httpx.Response:
+                return await client.get(
                     f"{API_BASE}/harvester/view_balance",
                     headers=headers,
                 )
 
-                if resp.status_code == 401:
-                    await self._refresh(client)
-                    headers = {"X-API-Key": self._access_token}
-                    resp = await client.get(
-                        f"{API_BASE}/harvester/view_balance",
-                        headers=headers,
-                    )
+            resp = await self._retry(_fetch_balance)
 
-                resp.raise_for_status()
-                data = resp.json()
+            if resp.status_code == 401:
+                await self._refresh(client)
+                headers = {"X-API-Key": self._access_token}
+                resp = await self._retry(_fetch_balance)
 
-                balance = float(data.get("data", {}).get("totalBalance", 0))
+            resp.raise_for_status()
+            data = resp.json()
 
-                return EarningsResult(
-                    platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
-                )
+            balance = float(data.get("data", {}).get("totalBalance", 0))
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("EarnFM collection failed: %s", exc)
+            logger.error("EarnFM collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/grass.py b/app/collectors/grass.py
index 7e41ef4..dfea661 100644
--- a/app/collectors/grass.py
+++ b/app/collectors/grass.py
@@ -25,6 +25,7 @@
 
 import asyncio
 import logging
+from enum import Enum
 
 import httpx
 
@@ -38,12 +39,18 @@
 _BASE_POINTS_PER_HOUR = 50
 
 
+class _GrassStatus(Enum):
+    AUTH_FAILED = "auth_failed"
+    RATE_LIMITED = "rate_limited"
+
+
 class GrassCollector(BaseCollector):
     """Collect earnings from Grass's API using an access token."""
 
     platform = "grass"
 
     def __init__(self, access_token: str) -> None:
+        super().__init__()
         self.access_token = access_token
 
     def _make_headers(self) -> dict[str, str]:
@@ -70,16 +77,16 @@ async def _request_with_retry(self, client: httpx.AsyncClient, url: str, **kwarg
                 await asyncio.sleep(wait)
         return resp  # Return last 429 response if all retries exhausted
 
-    async def _get_settled_points(self, client: httpx.AsyncClient) -> float:
+    async def _get_settled_points(self, client: httpx.AsyncClient) -> float | _GrassStatus:
         """Fetch totalPoints from /retrieveUser (non-zero only after epoch settles)."""
         resp = await self._request_with_retry(client, f"{API_BASE}/retrieveUser")
 
         if resp.status_code in (401, 403):
-            return -1.0  # Signal auth failure
+            return _GrassStatus.AUTH_FAILED
 
         if resp.status_code == 429:
             logger.warning("Grass API still rate-limited after retries")
-            return -2.0  # Signal rate limit
+            return _GrassStatus.RATE_LIMITED
 
         resp.raise_for_status()
         data = resp.json()
@@ -145,49 +152,49 @@ async def collect(self) -> EarningsResult:
         3. Otherwise, estimate from /devices uptime data.
         """
         try:
-            async with httpx.AsyncClient(timeout=60) as client:
-                # First, check if we have settled points
-                settled = await self._get_settled_points(client)
+            client = self._get_client(timeout=60)
 
-                if settled == -1.0:
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="Token expired — get a new accessToken from app.grass.io Local Storage",
-                    )
+            # First, check if we have settled points
+            settled = await self._get_settled_points(client)
 
-                if settled == -2.0:
+            if isinstance(settled, _GrassStatus):
+                if settled is _GrassStatus.AUTH_FAILED:
                     return EarningsResult(
                         platform=self.platform,
                         balance=0.0,
-                        error="Cloudflare rate limit — will retry next collection cycle",
-                    )
-
-                if settled > 0:
-                    # Epoch has settled, use the official total
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=round(settled, 4),
-                        currency="GRASS",
+                        error="Token expired — get a new accessToken from app.grass.io Local Storage",
                     )
+                return EarningsResult(
+                    platform=self.platform,
+                    balance=0.0,
+                    error="Cloudflare rate limit — will retry next collection cycle",
+                )
 
-                # Active epoch: estimate from active device uptime
-                estimated = await self._estimate_from_active_devices(client)
+            if settled > 0:
+                # Epoch has settled, use the official total
+                return EarningsResult(
+                    platform=self.platform,
+                    balance=round(settled, 4),
+                    currency="GRASS",
+                )
 
-                if estimated == -1.0:
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="Cloudflare rate limit — will retry next collection cycle",
-                    )
+            # Active epoch: estimate from active device uptime
+            estimated = await self._estimate_from_active_devices(client)
 
+            if estimated == -1.0:
                 return EarningsResult(
                     platform=self.platform,
-                    balance=round(estimated, 4),
-                    currency="GRASS",
+                    balance=0.0,
+                    error="Cloudflare rate limit — will retry next collection cycle",
                 )
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(estimated, 4),
+                currency="GRASS",
+            )
         except Exception as exc:
-            logger.error("Grass collection failed: %s", exc)
+            logger.error("Grass collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/honeygain.py b/app/collectors/honeygain.py
index 7e71f21..e8bfc60 100644
--- a/app/collectors/honeygain.py
+++ b/app/collectors/honeygain.py
@@ -21,8 +21,10 @@ class HoneygainCollector(BaseCollector):
     """Collect earnings from Honeygain's REST API."""
 
     platform = "honeygain"
+    _HEADERS = {"User-Agent": "Mozilla/5.0", "Referer": "https://dashboard.honeygain.com/"}
 
     def __init__(self, email: str, password: str) -> None:
+        super().__init__()
         self.email = email
         self.password = password
         self._token: str | None = None
@@ -32,10 +34,7 @@ async def _authenticate(self, client: httpx.AsyncClient) -> str:
         resp = await client.post(
             f"{API_BASE}/v1/users/tokens",
             json={"email": self.email, "password": self.password},
-            headers={
-                "User-Agent": "Mozilla/5.0",
-                "Referer": "https://dashboard.honeygain.com/",
-            },
+            headers=self._HEADERS,
         )
         resp.raise_for_status()
         data = resp.json()
@@ -47,11 +46,13 @@ async def _authenticate(self, client: httpx.AsyncClient) -> str:
     async def collect(self) -> EarningsResult:
         """Fetch current Honeygain balance."""
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
-                if not self._token:
-                    self._token = await self._authenticate(client)
+            client = self._get_client(timeout=30)
+
+            if not self._token:
+                self._token = await self._authenticate(client)
 
-                headers = {"Authorization": f"Bearer {self._token}"}
+            async def _fetch_balance():
+                headers = {**self._HEADERS, "Authorization": f"Bearer {self._token}"}
                 resp = await client.get(
                     f"{API_BASE}/v1/users/balances",
                     headers=headers,
@@ -60,27 +61,30 @@ async def collect(self) -> EarningsResult:
                 # Token may have expired — retry once
                 if resp.status_code == 401:
                     self._token = await self._authenticate(client)
-                    headers = {"Authorization": f"Bearer {self._token}"}
+                    headers = {**self._HEADERS, "Authorization": f"Bearer {self._token}"}
                     resp = await client.get(
                         f"{API_BASE}/v1/users/balances",
                         headers=headers,
                     )
 
                 resp.raise_for_status()
-                data = resp.json()
+                return resp
 
-                # Balance is in cents (usd_cents)
-                payout = data.get("data", {}).get("payout", {})
-                usd_cents = float(payout.get("usd_cents", 0))
-                balance_usd = round(usd_cents / 100, 4)
+            resp = await self._retry(_fetch_balance)
+            data = resp.json()
 
-                return EarningsResult(
-                    platform=self.platform,
-                    balance=balance_usd,
-                    currency="USD",
-                )
+            # Balance is in cents (usd_cents)
+            payout = data.get("data", {}).get("payout", {})
+            usd_cents = float(payout.get("usd_cents", 0))
+            balance_usd = round(usd_cents / 100, 4)
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=balance_usd,
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("Honeygain collection failed: %s", exc)
+            logger.error("Honeygain collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/iproyal.py b/app/collectors/iproyal.py
index 09eab84..caa70bb 100644
--- a/app/collectors/iproyal.py
+++ b/app/collectors/iproyal.py
@@ -31,15 +31,18 @@ class IPRoyalCollector(BaseCollector):
     platform = "iproyal"
 
     def __init__(self, email: str, password: str) -> None:
+        super().__init__()
         self.email = email
         self.password = password
+        self._device_id = _generate_identifier()
+        self._token: str | None = None
 
     async def _login(self, client: httpx.AsyncClient) -> str | None:
         """Login and return a JWT access token, or None on failure."""
         resp = await client.post(
             f"{API_BASE}/users/tokens",
             json={
-                "identifier": _generate_identifier(),
+                "identifier": self._device_id,
                 "email": self.email,
                 "password": self.password,
                 "h_captcha_response": "",
@@ -57,41 +60,51 @@ async def _login(self, client: httpx.AsyncClient) -> str | None:
         data = resp.json()
         return data.get("access_token")
 
+    async def _fetch_balance(self, client: httpx.AsyncClient) -> httpx.Response:
+        """Fetch balance dashboard (used as retry target)."""
+        return await client.get(
+            f"{API_BASE}/users/me/balance-dashboard",
+            headers={"Authorization": f"Bearer {self._token}"},
+        )
+
     async def collect(self) -> EarningsResult:
         """Fetch current IPRoyal Pawns balance."""
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
-                token = await self._login(client)
-                if not token:
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="Login failed — check email/password",
-                    )
+            client = self._get_client(timeout=30)
 
-                resp = await client.get(
-                    f"{API_BASE}/users/me/balance-dashboard",
-                    headers={"Authorization": f"Bearer {token}"},
+            if not self._token:
+                self._token = await self._login(client)
+            if not self._token:
+                return EarningsResult(
+                    platform=self.platform,
+                    balance=0.0,
+                    error="Login failed — check email/password",
                 )
 
-                if resp.status_code == 401:
+            resp = await self._retry(lambda: self._fetch_balance(client))
+
+            if resp.status_code == 401:
+                # Token expired, re-login once
+                self._token = await self._login(client)
+                if not self._token:
                     return EarningsResult(
                         platform=self.platform,
                         balance=0.0,
-                        error="Authentication expired",
+                        error="Re-login failed after 401",
                     )
+                resp = await self._fetch_balance(client)
 
-                resp.raise_for_status()
-                data = resp.json()
-                balance = float(data.get("balance", 0))
+            resp.raise_for_status()
+            data = resp.json()
+            balance = float(data.get("balance", 0))
 
-                return EarningsResult(
-                    platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
-                )
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("IPRoyal collection failed: %s", exc)
+            logger.error("IPRoyal collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/mystnodes.py b/app/collectors/mystnodes.py
index 1f9e17c..a18fc3a 100644
--- a/app/collectors/mystnodes.py
+++ b/app/collectors/mystnodes.py
@@ -28,6 +28,7 @@ def __init__(
         email: str = "",
         password: str = "",
     ) -> None:
+        super().__init__()
         self.email = email
         self.password = password
         self._access_token: str | None = None
@@ -92,21 +93,21 @@ async def collect(self) -> EarningsResult:
                 error="MystNodes email/password not configured",
             )
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
-                resp = await self._get_with_retry(client, f"{CLOUD_BASE}/node/total-earnings")
-                resp.raise_for_status()
-                data = resp.json()
-
-                # earningsTotal is in MYST tokens (Polygon)
-                total_myst = float(data.get("earningsTotal", 0))
-
-                return EarningsResult(
-                    platform=self.platform,
-                    balance=round(total_myst, 4),
-                    currency="MYST",
-                )
+            client = self._get_client(timeout=30)
+            resp = await self._retry(lambda: self._get_with_retry(client, f"{CLOUD_BASE}/node/total-earnings"))
+            resp.raise_for_status()
+            data = resp.json()
+
+            # earningsTotal is in MYST tokens (Polygon)
+            total_myst = float(data.get("earningsTotal", 0))
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(total_myst, 4),
+                currency="MYST",
+            )
         except Exception as exc:
-            logger.error("MystNodes collection failed: %s", exc)
+            logger.error("MystNodes collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
@@ -129,40 +130,42 @@ async def get_per_node_earnings(self) -> list[dict[str, Any]]:
         if not self.email or not self.password:
             return []
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
-                resp = await self._get_with_retry(
+            client = self._get_client(timeout=30)
+            resp = await self._retry(
+                lambda: self._get_with_retry(
                     client,
                     f"{CLOUD_BASE}/node",
                     params={"page": 1, "itemsPerPage": 100},
                 )
-                resp.raise_for_status()
-                data = resp.json()
-
-                result = []
-                for node in data.get("nodes", []):
-                    # Sum 30-day earnings across all service types
-                    earnings_30d = sum(float(e.get("etherAmount", 0)) for e in node.get("earnings", []))
-
-                    lifetime = node.get("lifetimeEarnings") or {}
-                    total_ether = float(lifetime.get("totalEther", 0))
-                    settled = float(lifetime.get("settledEther", 0))
-                    unsettled = float(lifetime.get("unsettledEther", 0))
-
-                    result.append(
-                        {
-                            "identity": node.get("identity", ""),
-                            "name": node.get("name") or "",
-                            "local_ip": node.get("localIp", ""),
-                            "online": (node.get("nodeStatus", {}).get("online", False)),
-                            "country": (node.get("country", {}).get("code", "")),
-                            "version": node.get("version", ""),
-                            "earnings_myst": round(earnings_30d, 6),
-                            "lifetime_myst": round(total_ether, 6),
-                            "lifetime_settled_myst": round(settled, 6),
-                            "lifetime_unsettled_myst": round(unsettled, 6),
-                        }
-                    )
-                return result
+            )
+            resp.raise_for_status()
+            data = resp.json()
+
+            result = []
+            for node in data.get("nodes", []):
+                # Sum 30-day earnings across all service types
+                earnings_30d = sum(float(e.get("etherAmount", 0)) for e in node.get("earnings", []))
+
+                lifetime = node.get("lifetimeEarnings") or {}
+                total_ether = float(lifetime.get("totalEther", 0))
+                settled = float(lifetime.get("settledEther", 0))
+                unsettled = float(lifetime.get("unsettledEther", 0))
+
+                result.append(
+                    {
+                        "identity": node.get("identity", ""),
+                        "name": node.get("name") or "",
+                        "local_ip": node.get("localIp", ""),
+                        "online": (node.get("nodeStatus", {}).get("online", False)),
+                        "country": (node.get("country", {}).get("code", "")),
+                        "version": node.get("version", ""),
+                        "earnings_myst": round(earnings_30d, 6),
+                        "lifetime_myst": round(total_ether, 6),
+                        "lifetime_settled_myst": round(settled, 6),
+                        "lifetime_unsettled_myst": round(unsettled, 6),
+                    }
+                )
+            return result
         except Exception as exc:
-            logger.error("MystNodes per-node fetch failed: %s", exc)
+            logger.error("MystNodes per-node fetch failed: %s", exc, exc_info=True)
             return []
diff --git a/app/collectors/packetstream.py b/app/collectors/packetstream.py
index 408fc94..e0d9856 100644
--- a/app/collectors/packetstream.py
+++ b/app/collectors/packetstream.py
@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import re
 
@@ -24,77 +25,79 @@ class PacketStreamCollector(BaseCollector):
     platform = "packetstream"
 
     def __init__(self, auth_token: str) -> None:
+        super().__init__()
         self.auth_token = auth_token
 
     async def collect(self) -> EarningsResult:
         """Fetch current PacketStream balance by scraping dashboard."""
         try:
             cookies = {"auth": self.auth_token}
+            client = self._get_client(cookies=cookies)
 
-            async with httpx.AsyncClient(timeout=30, cookies=cookies) as client:
-                resp = await client.get(f"{API_BASE}/dashboard")
+            async def _fetch() -> httpx.Response:
+                return await client.get(f"{API_BASE}/dashboard")
 
-                if resp.status_code in (401, 403) or "/login" in str(resp.url):
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="Authentication failed — check auth JWT cookie",
-                    )
+            resp = await self._retry(_fetch)
 
-                resp.raise_for_status()
-                html = resp.text
+            if resp.status_code in (401, 403) or "/login" in str(resp.url):
+                return EarningsResult(
+                    platform=self.platform,
+                    balance=0.0,
+                    error="Authentication failed — check auth JWT cookie",
+                )
+
+            resp.raise_for_status()
+            html = resp.text
 
-                balance = 0.0
-                parsed = False
+            balance = 0.0
+            parsed = False
 
-                # Pattern 1: server-rendered Balance card (current)
-                # <h3>Balance</h3>...<h2 ...>$0.13</h2>
+            # Pattern 1: server-rendered Balance card (current)
+            # <h3>Balance</h3>...<h2 ...>$0.13</h2>
+            match = re.search(
+                r"<h3>Balance</h3>.*?<h2[^>]*>\$?([\d.]+)</h2>",
+                html,
+                re.DOTALL,
+            )
+            if match:
+                balance = float(match.group(1))
+                parsed = True
+
+            # Pattern 2: window.userData JSON (legacy)
+            if not parsed:
                 match = re.search(
-                    r"<h3>Balance</h3>.*?<h2[^>]*>\$?([\d.]+)</h2>",
+                    r"window\.userData\s*=\s*(\{[^}]+\})",
                     html,
-                    re.DOTALL,
                 )
                 if match:
-                    balance = float(match.group(1))
-                    parsed = True
-
-                # Pattern 2: window.userData JSON (legacy)
-                if not parsed:
-                    match = re.search(
-                        r"window\.userData\s*=\s*(\{[^}]+\})",
-                        html,
-                    )
-                    if match:
-                        import json
-
-                        try:
-                            user_data = json.loads(match.group(1))
-                            balance = float(user_data.get("balance", 0))
-                            parsed = True
-                        except (json.JSONDecodeError, ValueError):
-                            pass
-
-                # Pattern 3: bare "balance" key in JSON
-                if not parsed:
-                    match = re.search(r'"balance"\s*:\s*([\d.]+)', html)
-                    if match:
-                        balance = float(match.group(1))
+                    try:
+                        user_data = json.loads(match.group(1))
+                        balance = float(user_data.get("balance", 0))
                         parsed = True
+                    except (json.JSONDecodeError, ValueError):
+                        pass
 
-                if not parsed:
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="Could not parse balance from dashboard — page structure may have changed",
-                    )
+            # Pattern 3: bare "balance" key in JSON
+            if not parsed:
+                match = re.search(r'"balance"\s*:\s*([\d.]+)', html)
+                if match:
+                    balance = float(match.group(1))
+                    parsed = True
 
+            if not parsed:
                 return EarningsResult(
                     platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
+                    balance=0.0,
+                    error="Could not parse balance from dashboard — page structure may have changed",
                 )
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("PacketStream collection failed: %s", exc)
+            logger.error("PacketStream collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/proxyrack.py b/app/collectors/proxyrack.py
index 36fe17b..1273356 100644
--- a/app/collectors/proxyrack.py
+++ b/app/collectors/proxyrack.py
@@ -8,7 +8,7 @@
 
 import logging
 
-import httpx
+import httpx  # noqa: F401 — needed for test patching
 
 from app.collectors.base import BaseCollector, EarningsResult
 
@@ -23,6 +23,7 @@ class ProxyRackCollector(BaseCollector):
     platform = "proxyrack"
 
     def __init__(self, api_key: str) -> None:
+        super().__init__()
         self.api_key = api_key
 
     async def collect(self) -> EarningsResult:
@@ -35,7 +36,8 @@ async def collect(self) -> EarningsResult:
                 "Content-Type": "application/json",
             }
 
-            async with httpx.AsyncClient(timeout=30) as client:
+            async def _fetch_balance():
+                client = self._get_client(timeout=30)
                 resp = await client.post(
                     f"{API_BASE}/balance",
                     headers=headers,
@@ -60,8 +62,10 @@ async def collect(self) -> EarningsResult:
                     balance=round(balance, 4),
                     currency="USD",
                 )
+
+            return await self._retry(_fetch_balance)
         except Exception as exc:
-            logger.error("ProxyRack collection failed: %s", exc)
+            logger.error("ProxyRack collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/repocket.py b/app/collectors/repocket.py
index 9b0ab18..1883a4f 100644
--- a/app/collectors/repocket.py
+++ b/app/collectors/repocket.py
@@ -26,6 +26,7 @@ class RepocketCollector(BaseCollector):
     platform = "repocket"
 
     def __init__(self, email: str, password: str) -> None:
+        super().__init__()
         self.email = email
         self.password = password
         self._id_token: str | None = None
@@ -71,38 +72,39 @@ async def _refresh(self, client: httpx.AsyncClient) -> str:
     async def collect(self) -> EarningsResult:
         """Fetch current Repocket balance."""
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
-                if not self._id_token:
-                    await self._authenticate(client)
+            client = self._get_client(timeout=30)
+            if not self._id_token:
+                await self._authenticate(client)
 
-                headers = {"Auth-Token": self._id_token}
-                resp = await client.get(
+            headers = {"Auth-Token": self._id_token}
+
+            async def _fetch_balance() -> httpx.Response:
+                return await client.get(
                     f"{API_BASE}/reports/current",
                     headers=headers,
                 )
 
-                if resp.status_code == 401:
-                    await self._refresh(client)
-                    headers = {"Auth-Token": self._id_token}
-                    resp = await client.get(
-                        f"{API_BASE}/reports/current",
-                        headers=headers,
-                    )
-
-                resp.raise_for_status()
-                data = resp.json()
-
-                # centsCredited is in cents
-                cents = float(data.get("centsCredited", 0))
-                balance_usd = round(cents / 100, 4)
-
-                return EarningsResult(
-                    platform=self.platform,
-                    balance=balance_usd,
-                    currency="USD",
-                )
+            resp = await self._retry(_fetch_balance)
+
+            if resp.status_code == 401:
+                await self._refresh(client)
+                headers = {"Auth-Token": self._id_token}
+                resp = await self._retry(_fetch_balance)
+
+            resp.raise_for_status()
+            data = resp.json()
+
+            # centsCredited is in cents
+            cents = float(data.get("centsCredited", 0))
+            balance_usd = round(cents / 100, 4)
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=balance_usd,
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("Repocket collection failed: %s", exc)
+            logger.error("Repocket collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/salad.py b/app/collectors/salad.py
index f22143c..b080c5b 100644
--- a/app/collectors/salad.py
+++ b/app/collectors/salad.py
@@ -29,6 +29,7 @@ class SaladCollector(BaseCollector):
     platform = "salad"
 
     def __init__(self, auth_cookie: str) -> None:
+        super().__init__()
         self.auth_cookie = auth_cookie
 
     async def collect(self) -> EarningsResult:
@@ -36,33 +37,35 @@ async def collect(self) -> EarningsResult:
         try:
             cookies = {"auth": self.auth_cookie}
             headers = {"X-XSRF-TOKEN": self.auth_cookie}
+            client = self._get_client(cookies=cookies)
 
-            async with httpx.AsyncClient(timeout=30) as client:
-                resp = await client.get(
+            async def _fetch() -> httpx.Response:
+                return await client.get(
                     f"{API_BASE}/profile/balance",
-                    cookies=cookies,
                     headers=headers,
                 )
 
-                if resp.status_code in (401, 403):
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="Auth cookie expired — get a new 'auth' cookie from salad.com",
-                    )
-
-                resp.raise_for_status()
-                data = resp.json()
-
-                balance = float(data.get("currentBalance", 0))
+            resp = await self._retry(_fetch)
 
+            if resp.status_code in (401, 403):
                 return EarningsResult(
                     platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
+                    balance=0.0,
+                    error="Auth cookie expired — get a new 'auth' cookie from salad.com",
                 )
+
+            resp.raise_for_status()
+            data = resp.json()
+
+            balance = float(data.get("currentBalance", 0))
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("Salad collection failed: %s", exc)
+            logger.error("Salad collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/storj.py b/app/collectors/storj.py
index 7b995e2..91b2b20 100644
--- a/app/collectors/storj.py
+++ b/app/collectors/storj.py
@@ -23,41 +23,42 @@ class StorjCollector(BaseCollector):
     platform = "storj"
 
     def __init__(self, api_url: str = DEFAULT_API_URL) -> None:
+        super().__init__()
         self.api_url = api_url.rstrip("/")
 
     async def collect(self) -> EarningsResult:
         """Fetch current Storj storagenode estimated payout."""
         try:
-            async with httpx.AsyncClient(timeout=15) as client:
-                resp = await client.get(f"{self.api_url}/api/sno/estimated-payout")
-
-                if resp.status_code == 404:
-                    # Older storagenode versions use different endpoint
-                    resp = await client.get(f"{self.api_url}/api/sno")
-
-                resp.raise_for_status()
-                data = resp.json()
-
-                # estimated-payout endpoint returns cents
-                if "currentMonth" in data:
-                    # Payout values are in cents (integer)
-                    payout_cents = data["currentMonth"].get("egressBandwidthPayout", 0)
-                    payout_cents += data["currentMonth"].get("egressRepairAuditPayout", 0)
-                    payout_cents += data["currentMonth"].get("diskSpacePayout", 0)
-                    balance = payout_cents / 100.0
-                elif "estimatedPayout" in data:
-                    balance = data["estimatedPayout"] / 100.0
-                elif "currentMonthExpectations" in data:
-                    balance = data["currentMonthExpectations"] / 100.0
-                else:
-                    # Fallback: try /api/sno for totalPayout
-                    balance = 0.0
-
-                return EarningsResult(
-                    platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
-                )
+            client = self._get_client(timeout=15)
+            resp = await self._retry(lambda: client.get(f"{self.api_url}/api/sno/estimated-payout"))
+
+            if resp.status_code == 404:
+                # Older storagenode versions use different endpoint
+                resp = await client.get(f"{self.api_url}/api/sno")
+
+            resp.raise_for_status()
+            data = resp.json()
+
+            # estimated-payout endpoint returns cents
+            if "currentMonth" in data:
+                # Payout values are in cents (integer)
+                payout_cents = data["currentMonth"].get("egressBandwidthPayout", 0)
+                payout_cents += data["currentMonth"].get("egressRepairAuditPayout", 0)
+                payout_cents += data["currentMonth"].get("diskSpacePayout", 0)
+                balance = payout_cents / 100.0
+            elif "estimatedPayout" in data:
+                balance = data["estimatedPayout"] / 100.0
+            elif "currentMonthExpectations" in data:
+                balance = data["currentMonthExpectations"] / 100.0
+            else:
+                # Fallback: try /api/sno for totalPayout
+                balance = 0.0
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+            )
         except httpx.ConnectError:
             return EarningsResult(
                 platform=self.platform,
@@ -70,7 +71,7 @@ async def collect(self) -> EarningsResult:
                 ),
             )
         except Exception as exc:
-            logger.error("Storj collection failed: %s", exc)
+            logger.error("Storj collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/collectors/traffmonetizer.py b/app/collectors/traffmonetizer.py
index 0db47eb..4c5832e 100644
--- a/app/collectors/traffmonetizer.py
+++ b/app/collectors/traffmonetizer.py
@@ -13,7 +13,7 @@
 
 import logging
 
-import httpx
+import httpx  # noqa: F401 (used by test patches targeting this module)
 
 from app.collectors.base import BaseCollector, EarningsResult
 
@@ -28,6 +28,7 @@ class TraffmonetizerCollector(BaseCollector):
     platform = "traffmonetizer"
 
     def __init__(self, token: str = "") -> None:
+        super().__init__()
         self._token = token.strip()
 
     async def collect(self) -> EarningsResult:
@@ -40,33 +41,35 @@ async def collect(self) -> EarningsResult:
             )
 
         try:
-            async with httpx.AsyncClient(timeout=30) as client:
-                headers = {"Authorization": f"Bearer {self._token}"}
+            client = self._get_client(timeout=30)
+            headers = {"Authorization": f"Bearer {self._token}"}
 
-                resp = await client.get(
+            resp = await self._retry(
+                lambda: client.get(
                     f"{API_BASE}/app_user/get_balance",
                     headers=headers,
                 )
+            )
 
-                if resp.status_code in (401, 403):
-                    return EarningsResult(
-                        platform=self.platform,
-                        balance=0.0,
-                        error="Token expired — refresh access_token from browser Local Storage",
-                    )
-
-                resp.raise_for_status()
-                data = resp.json()
-
-                balance = float(data.get("data", {}).get("balance", 0))
-
+            if resp.status_code in (401, 403):
                 return EarningsResult(
                     platform=self.platform,
-                    balance=round(balance, 4),
-                    currency="USD",
+                    balance=0.0,
+                    error="Token expired — refresh access_token from browser Local Storage",
                 )
+
+            resp.raise_for_status()
+            data = resp.json()
+
+            balance = float(data.get("data", {}).get("balance", 0))
+
+            return EarningsResult(
+                platform=self.platform,
+                balance=round(balance, 4),
+                currency="USD",
+            )
         except Exception as exc:
-            logger.error("Traffmonetizer collection failed: %s", exc)
+            logger.error("Traffmonetizer collection failed: %s", exc, exc_info=True)
             return EarningsResult(
                 platform=self.platform,
                 balance=0.0,
diff --git a/app/database.py b/app/database.py
index cc50a35..6296e95 100644
--- a/app/database.py
+++ b/app/database.py
@@ -213,6 +213,13 @@ async def init_db() -> None:
             """)
         elif "apps" not in cols:
             await db.execute("ALTER TABLE workers ADD COLUMN apps TEXT NOT NULL DEFAULT '[]'")
+
+        # Migrate users table: add password_changed_at for session invalidation
+        cursor = await db.execute("PRAGMA table_info(users)")
+        user_cols = {row["name"] for row in await cursor.fetchall()}
+        if "password_changed_at" not in user_cols:
+            await db.execute("ALTER TABLE users ADD COLUMN password_changed_at REAL DEFAULT 0")
+
         await db.commit()
     finally:
         await db.close()
@@ -689,6 +696,21 @@ async def delete_user(user_id: int) -> None:
         await db.close()
 
 
+async def update_user_password(user_id: int, hashed_password: str) -> None:
+    """Update a user's password and record the change timestamp."""
+    import time
+
+    db = await _get_db()
+    try:
+        await db.execute(
+            "UPDATE users SET password = ?, password_changed_at = ? WHERE id = ?",
+            (hashed_password, time.time(), user_id),
+        )
+        await db.commit()
+    finally:
+        await db.close()
+
+
 # --- User Preferences ---
 
 
diff --git a/app/main.py b/app/main.py
index 08a3402..fbcde0a 100644
--- a/app/main.py
+++ b/app/main.py
@@ -187,9 +187,10 @@ async def _run_collection() -> None:
             config = await database.get_config() or {}
             if not isinstance(config, dict):
                 config = {}
-            from app.collectors import make_collectors
+            from app.collectors import _close_stale, make_collectors
 
             collectors = make_collectors(deployments, config)
+            await _close_stale()
             results = await asyncio.gather(*(c.collect() for c in collectors), return_exceptions=True)
             alerts: list[dict[str, str]] = []
             for result in results:
@@ -267,6 +268,9 @@ async def lifespan(app: FastAPI):
 
     # Shutdown
     scheduler.shutdown(wait=False)
+    from app.collectors import close_all_collectors
+
+    await close_all_collectors()
     logger.info("CashPilot stopped")
 
 
@@ -488,7 +492,7 @@ async def do_register(
             status_code=400,
         )
 
-    if len(password) < 8:
+    if len(password) < 10:
         return templates.TemplateResponse(
             request,
             "auth.html",
@@ -498,7 +502,7 @@ async def do_register(
                 "mode": "register",
                 "action": "/register",
                 "button_text": "Create Account",
-                "error": "Password must be at least 8 characters",
+                "error": "Password must be at least 10 characters",
                 "is_first": is_first,
             },
             status_code=400,
@@ -1286,11 +1290,21 @@ async def api_collect(request: Request) -> dict[str, str]:
     return {"status": "collection_started"}
 
 
+_MAX_ALERT_ERROR_LEN = 200
+
+
 @app.get("/api/collector-alerts")
 async def api_collector_alerts(request: Request) -> list[dict[str, str]]:
-    """Return collector errors from the last collection run."""
+    """Return collector errors from the last collection run (sanitized)."""
     _require_auth_api(request)
-    return _collector_alerts
+    sanitized: list[dict[str, str]] = []
+    for alert in _collector_alerts:
+        error_msg = alert.get("error", "")
+        clean = error_msg[:_MAX_ALERT_ERROR_LEN]
+        if len(error_msg) > _MAX_ALERT_ERROR_LEN:
+            clean += "..."
+        sanitized.append({"platform": alert["platform"], "error": clean})
+    return sanitized
 
 
 @app.get("/api/exchange-rates")
diff --git a/app/worker_api.py b/app/worker_api.py
index 9386963..f7032ff 100644
--- a/app/worker_api.py
+++ b/app/worker_api.py
@@ -81,7 +81,7 @@ async def _send_heartbeat() -> None:
 
     containers = []
     try:
-        containers = orchestrator.get_status()
+        containers = await asyncio.to_thread(orchestrator.get_status)
     except Exception as exc:
         logger.warning("Failed to get container status for heartbeat: %s", exc)
 
@@ -180,7 +180,7 @@ async def worker_status_page():
     """Self-contained HTML status page for the worker."""
     containers = []
     try:
-        containers = orchestrator.get_status_cached()
+        containers = await asyncio.to_thread(orchestrator.get_status_cached)
     except Exception as exc:
         logger.warning("Failed to get container status for status page: %s", exc)
 
@@ -293,7 +293,7 @@ async def api_worker_status(request: Request) -> dict[str, Any]:
     _verify_api_key(request)
     containers = []
     try:
-        containers = orchestrator.get_status_cached()
+        containers = await asyncio.to_thread(orchestrator.get_status_cached)
     except Exception as exc:
         logger.warning("Failed to get container status: %s", exc)
     return {
@@ -310,7 +310,7 @@ async def api_list_containers(request: Request) -> list[dict[str, Any]]:
     """List all CashPilot-managed containers."""
     _verify_api_key(request)
     try:
-        return orchestrator.get_status_cached()
+        return await asyncio.to_thread(orchestrator.get_status_cached)
     except RuntimeError as exc:
         raise HTTPException(status_code=503, detail=str(exc))
 
@@ -321,7 +321,8 @@ async def api_deploy_container(request: Request, slug: str, spec: DeploySpec) ->
     _verify_api_key(request)
     _validate_deploy_spec(spec)
     try:
-        container_id = orchestrator.deploy_raw(
+        container_id = await asyncio.to_thread(
+            orchestrator.deploy_raw,
             slug=slug,
             image=spec.image,
             env=spec.env,
@@ -344,7 +345,7 @@ async def api_deploy_container(request: Request, slug: str, spec: DeploySpec) ->
 async def api_restart_container(request: Request, slug: str) -> dict[str, str]:
     _verify_api_key(request)
     try:
-        orchestrator.restart_service(slug)
+        await asyncio.to_thread(orchestrator.restart_service, slug)
         return {"status": "restarted"}
     except ValueError as exc:
         raise HTTPException(status_code=404, detail=str(exc))
@@ -356,7 +357,7 @@ async def api_restart_container(request: Request, slug: str) -> dict[str, str]:
 async def api_stop_container(request: Request, slug: str) -> dict[str, str]:
     _verify_api_key(request)
     try:
-        orchestrator.stop_service(slug)
+        await asyncio.to_thread(orchestrator.stop_service, slug)
         return {"status": "stopped"}
     except ValueError as exc:
         raise HTTPException(status_code=404, detail=str(exc))
@@ -368,7 +369,7 @@ async def api_stop_container(request: Request, slug: str) -> dict[str, str]:
 async def api_start_container(request: Request, slug: str) -> dict[str, str]:
     _verify_api_key(request)
     try:
-        orchestrator.start_service(slug)
+        await asyncio.to_thread(orchestrator.start_service, slug)
         return {"status": "started"}
     except ValueError as exc:
         raise HTTPException(status_code=404, detail=str(exc))
@@ -380,7 +381,7 @@ async def api_start_container(request: Request, slug: str) -> dict[str, str]:
 async def api_remove_container(request: Request, slug: str, delete_volumes: bool = False) -> dict[str, Any]:
     _verify_api_key(request)
     try:
-        result = orchestrator.remove_service(slug, delete_volumes=delete_volumes)
+        result = await asyncio.to_thread(orchestrator.remove_service, slug, delete_volumes=delete_volumes)
         return {"status": "removed", **result}
     except ValueError as exc:
         raise HTTPException(status_code=404, detail=str(exc))
@@ -392,7 +393,7 @@ async def api_remove_container(request: Request, slug: str, delete_volumes: bool
 async def api_container_logs(request: Request, slug: str, lines: int = 50) -> dict[str, str]:
     _verify_api_key(request)
     try:
-        logs = orchestrator.get_service_logs(slug, lines=min(lines, 1000))
+        logs = await asyncio.to_thread(orchestrator.get_service_logs, slug, lines=min(lines, 1000))
         return {"logs": logs}
     except ValueError as exc:
         raise HTTPException(status_code=404, detail=str(exc))
diff --git a/tests/test_collectors.py b/tests/test_collectors.py
index 0c3a840..55da7fa 100644
--- a/tests/test_collectors.py
+++ b/tests/test_collectors.py
@@ -38,7 +38,6 @@ def test_earnings_result_fields():
     assert result.platform == "test"
     assert result.balance == 1.23
     assert result.currency == "USD"
-    assert result.bytes_uploaded == 0
     assert result.error is None
 
 
diff --git a/tests/test_collectors_deep.py b/tests/test_collectors_deep.py
index 90d3f92..d99f53c 100644
--- a/tests/test_collectors_deep.py
+++ b/tests/test_collectors_deep.py
@@ -400,7 +400,10 @@ def test_collect_api_fallback(self):
         from app.collectors.bytelixir import BytelixirCollector
 
         # HTML scrape fails (no balance pattern)
+        url_mock = MagicMock()
+        url_mock.path = "/en"
         html_resp = _mock_response(200, text="<p>no balance</p>", url="https://dash.bytelixir.com/en")
+        html_resp.url = url_mock
         html_resp.text = "<p>no balance</p>"
         # API fallback succeeds
         api_resp = _mock_response(200, {"data": {"balance": "0.5000000000"}})
@@ -408,7 +411,7 @@ def test_collect_api_fallback(self):
         client = _make_async_client()
         client.get.side_effect = [html_resp, api_resp]
 
-        with patch.object(BytelixirCollector, "_make_client", return_value=client):
+        with patch.object(BytelixirCollector, "_get_client", return_value=client):
             c = BytelixirCollector(session_cookie="valid")
             result = asyncio.run(c.collect())
         assert result.balance == 0.50
@@ -416,7 +419,10 @@ def test_collect_api_fallback(self):
     def test_collect_api_401(self):
         from app.collectors.bytelixir import BytelixirCollector
 
+        url_mock = MagicMock()
+        url_mock.path = "/en"
         html_resp = _mock_response(200, text="<p>no balance</p>", url="https://dash.bytelixir.com/en")
+        html_resp.url = url_mock
         html_resp.text = "<p>no balance</p>"
         api_resp = _mock_response(401)
         api_resp.raise_for_status = MagicMock()  # 401 handled inline
@@ -424,7 +430,7 @@ def test_collect_api_401(self):
         client = _make_async_client()
         client.get.side_effect = [html_resp, api_resp]
 
-        with patch.object(BytelixirCollector, "_make_client", return_value=client):
+        with patch.object(BytelixirCollector, "_get_client", return_value=client):
             c = BytelixirCollector(session_cookie="expired")
             result = asyncio.run(c.collect())
         assert result.error is not None
diff --git a/tests/test_collectors_extended.py b/tests/test_collectors_extended.py
index fedf804..2c05b1f 100644
--- a/tests/test_collectors_extended.py
+++ b/tests/test_collectors_extended.py
@@ -494,11 +494,14 @@ class TestBytelixirCollector:
     def test_collect_session_expired(self):
         from app.collectors.bytelixir import BytelixirCollector
 
+        url_mock = MagicMock()
+        url_mock.path = "/login"
         resp = _mock_response(200, url="https://dash.bytelixir.com/login")
+        resp.url = url_mock
         client = _make_async_client()
         client.get.return_value = resp
 
-        with patch.object(BytelixirCollector, "_make_client", return_value=client):
+        with patch.object(BytelixirCollector, "_get_client", return_value=client):
             c = BytelixirCollector(session_cookie="expired")
             result = asyncio.run(c.collect())
         assert result.error is not None
@@ -508,12 +511,15 @@ def test_collect_html_scrape_success(self):
         from app.collectors.bytelixir import BytelixirCollector
 
         html = '<span>$</span>0.04<span class="text-2xs">025</span>'
+        url_mock = MagicMock()
+        url_mock.path = "/en"
         resp = _mock_response(200, text=html, url="https://dash.bytelixir.com/en")
+        resp.url = url_mock
         resp.text = html
         client = _make_async_client()
         client.get.return_value = resp
 
-        with patch.object(BytelixirCollector, "_make_client", return_value=client):
+        with patch.object(BytelixirCollector, "_get_client", return_value=client):
             c = BytelixirCollector(session_cookie="valid-sess")
             result = asyncio.run(c.collect())
         assert result.error is None
@@ -525,7 +531,7 @@ def test_collect_error(self):
         client = _make_async_client()
         client.get.side_effect = Exception("Error")
 
-        with patch.object(BytelixirCollector, "_make_client", return_value=client):
+        with patch.object(BytelixirCollector, "_get_client", return_value=client):
             c = BytelixirCollector(session_cookie="bad")
             result = asyncio.run(c.collect())
         assert result.error is not None
diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py
index 1d204d6..1d0f753 100644
--- a/tests/test_coverage_gaps.py
+++ b/tests/test_coverage_gaps.py
@@ -218,14 +218,14 @@ def test_deployment_without_slug_key(self):
 
 
 # ---------------------------------------------------------------------------
-# bytelixir.py — _make_client, parse balance edge cases
+# bytelixir.py — _get_client, parse balance edge cases
 # ---------------------------------------------------------------------------
 
 
-class TestBytelixirMakeClient:
-    """Cover lines 59-77: _make_client with remember_web and xsrf_token."""
+class TestBytelixirGetClient:
+    """Cover _get_client with remember_web and xsrf_token."""
 
-    def test_make_client_with_all_cookies(self):
+    def test_get_client_with_all_cookies(self):
         from app.collectors.bytelixir import BytelixirCollector
 
         c = BytelixirCollector(
@@ -233,7 +233,7 @@ def test_make_client_with_all_cookies(self):
             remember_web="remember-val",
             xsrf_token="xsrf-val",
         )
-        client = c._make_client()
+        client = c._get_client()
         assert isinstance(client, httpx.AsyncClient)
         # Verify cookies are set
         cookies = dict(client.cookies)
@@ -242,11 +242,11 @@ def test_make_client_with_all_cookies(self):
         assert "XSRF-TOKEN" in cookies
         asyncio.run(client.aclose())
 
-    def test_make_client_session_only(self):
+    def test_get_client_session_only(self):
         from app.collectors.bytelixir import BytelixirCollector
 
         c = BytelixirCollector(session_cookie="sess-only")
-        client = c._make_client()
+        client = c._get_client()
         cookies = dict(client.cookies)
         assert "bytelixir_session" in cookies
         assert c._REMEMBER_COOKIE not in cookies

From 54394e2a266f98aa69edf5c557c1375c0055b2d7 Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 15:20:19 +0200
Subject: [PATCH 7/9] fix: update EarnApp API version and add Traffmonetizer
 CORS headers

EarnApp dashboard API now requires version 1.627.783. Traffmonetizer
needs Origin/Referer headers matching the browser app to avoid 403s.
---
 app/collectors/earnapp.py        | 2 +-
 app/collectors/traffmonetizer.py | 6 +++++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/app/collectors/earnapp.py b/app/collectors/earnapp.py
index e323bf5..700f12a 100644
--- a/app/collectors/earnapp.py
+++ b/app/collectors/earnapp.py
@@ -21,7 +21,7 @@ class EarnAppCollector(BaseCollector):
     """Collect earnings from EarnApp's dashboard API."""
 
     platform = "earnapp"
-    _API_VERSION = "1.613.719"
+    _API_VERSION = "1.627.783"
 
     def __init__(self, oauth_token: str, brd_sess_id: str = "") -> None:
         super().__init__()
diff --git a/app/collectors/traffmonetizer.py b/app/collectors/traffmonetizer.py
index 4c5832e..3c5077c 100644
--- a/app/collectors/traffmonetizer.py
+++ b/app/collectors/traffmonetizer.py
@@ -42,7 +42,11 @@ async def collect(self) -> EarningsResult:
 
         try:
             client = self._get_client(timeout=30)
-            headers = {"Authorization": f"Bearer {self._token}"}
+            headers = {
+                "Authorization": f"Bearer {self._token}",
+                "Origin": "https://app.traffmonetizer.com",
+                "Referer": "https://app.traffmonetizer.com/",
+            }
 
             resp = await self._retry(
                 lambda: client.get(

From af6e093a8a9acdd84fc5a897543825b58d229843 Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 15:29:05 +0200
Subject: [PATCH 8/9] fix: address CodeRabbit review findings

- Rate limiter only counts failed logins (not successful ones)
- Volume validation blocks subpaths of sensitive mounts
- Strengthen delete_volumes test assertion to check exact params
---
 app/main.py                      | 10 ++++++++--
 app/worker_api.py                |  6 +++++-
 tests/test_main_deploy_routes.py |  3 +--
 3 files changed, 14 insertions(+), 5 deletions(-)

diff --git a/app/main.py b/app/main.py
index fbcde0a..64e0c5d 100644
--- a/app/main.py
+++ b/app/main.py
@@ -56,7 +56,10 @@ def _check_login_rate(client_ip: str) -> None:
     _login_attempts[client_ip] = [t for t in attempts if now - t < _LOGIN_WINDOW_SECONDS]
     if len(_login_attempts[client_ip]) >= _LOGIN_MAX_ATTEMPTS:
         raise HTTPException(status_code=429, detail="Too many login attempts. Try again in a few minutes.")
-    _login_attempts[client_ip].append(now)
+
+
+def _record_failed_login(client_ip: str) -> None:
+    _login_attempts[client_ip].append(monotonic())
 
 
 def _safe_json(raw: str, fallback: Any = None) -> Any:
@@ -393,9 +396,11 @@ async def do_login(
     username: str = Form(...),
     password: str = Form(...),
 ):
-    _check_login_rate(request.client.host if request.client else "unknown")
+    client_ip = request.client.host if request.client else "unknown"
+    _check_login_rate(client_ip)
     user = await database.get_user_by_username(username)
     if not user or not auth.verify_password(password, user["password"]):
+        _record_failed_login(client_ip)
         return templates.TemplateResponse(
             request,
             "auth.html",
@@ -411,6 +416,7 @@ async def do_login(
             status_code=401,
         )
 
+    _login_attempts.pop(client_ip, None)
     token = auth.create_session_token(user["id"], user["username"], user["role"])
     response = RedirectResponse("/", status_code=303)
     return auth.set_session_cookie(response, token)
diff --git a/app/worker_api.py b/app/worker_api.py
index f7032ff..08bae69 100644
--- a/app/worker_api.py
+++ b/app/worker_api.py
@@ -283,8 +283,12 @@ def _validate_deploy_spec(spec: DeploySpec) -> None:
         if blocked:
             raise HTTPException(status_code=403, detail=f"Blocked capabilities: {', '.join(blocked)}")
     for source in spec.volumes:
-        if source.rstrip("/") in _BLOCKED_VOLUME_SOURCES or source == "/":
+        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 + "/"):
+                raise HTTPException(status_code=403, detail=f"Volume mount '{source}' is blocked")
 
 
 @app.get("/api/status")
diff --git a/tests/test_main_deploy_routes.py b/tests/test_main_deploy_routes.py
index 4d5af80..d258130 100644
--- a/tests/test_main_deploy_routes.py
+++ b/tests/test_main_deploy_routes.py
@@ -281,8 +281,7 @@ def test_remove_with_delete_volumes(self, client):
         ):
             resp = client.delete("/api/services/honeygain?delete_volumes=true")
             assert resp.status_code == 200
-            call_args = mock_client.delete.call_args
-            assert "delete_volumes" in str(call_args)
+            assert mock_client.delete.call_args.kwargs["params"] == {"delete_volumes": "true"}
 
 
 # ---------------------------------------------------------------------------

From 3db7077d99c4585e4c9d610438776dc97d54184e Mon Sep 17 00:00:00 2001
From: GeiserX <9169332+GeiserX@users.noreply.github.com>
Date: Tue, 19 May 2026 15:53:59 +0200
Subject: [PATCH 9/9] fix: wrap iproyal post-refresh balance fetch in _retry

Ensures transient network errors after token refresh get the same
retry/backoff protection as the initial fetch.
---
 app/collectors/iproyal.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/app/collectors/iproyal.py b/app/collectors/iproyal.py
index caa70bb..6ab403c 100644
--- a/app/collectors/iproyal.py
+++ b/app/collectors/iproyal.py
@@ -92,7 +92,7 @@ async def collect(self) -> EarningsResult:
                         balance=0.0,
                         error="Re-login failed after 401",
                     )
-                resp = await self._fetch_balance(client)
+                resp = await self._retry(lambda: self._fetch_balance(client))
 
             resp.raise_for_status()
             data = resp.json()