feat: security hardening, collector refactor, async orchestrator

app/auth.py

@@ -5,9 +5,11 @@
 
 from __future__ import annotations
 
+import hmac
 import logging
 import os
 import secrets
+import time
 from pathlib import Path
 from typing import Any
 
@@ -75,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)
@@ -88,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
 
@@ -108,10 +118,10 @@ 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

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",
 ]

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

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,