feat(fleet)!: per-worker fleet keys (full cutover, v1.0.0)

.github/workflows/release.yml

@@ -99,10 +99,29 @@ jobs:
           MAJOR=$(echo "$VERSION" | cut -d. -f1)
           MINOR=$(echo "$VERSION" | cut -d. -f2)
           PATCH=$(echo "$VERSION" | cut -d. -f3)
-          NEW_PATCH=$((PATCH + 1))
-          NEW_TAG="v${MAJOR}.${MINOR}.${NEW_PATCH}"
+
+          # Choose the bump from conventional-commit markers in the commits since
+          # the last tag: a "BREAKING CHANGE" footer or a "type!:" subject -> major;
+          # a "feat:" -> minor; otherwise patch. Still fully automated — no manual tags.
+          if [ "$LATEST" = "v0.0.0" ]; then
+            RANGE="HEAD"
+          else
+            RANGE="${LATEST}..HEAD"
+          fi
+          LOG=$(git log --format='%s%n%b' "$RANGE")
+          if echo "$LOG" | grep -qE '^BREAKING CHANGE:|^[a-z]+(\([^)]*\))?!:'; then
+            MAJOR=$((MAJOR + 1)); MINOR=0; PATCH=0
+            LEVEL="major"
+          elif echo "$LOG" | grep -qE '^feat(\([^)]*\))?:'; then
+            MINOR=$((MINOR + 1)); PATCH=0
+            LEVEL="minor"
+          else
+            PATCH=$((PATCH + 1))
+            LEVEL="patch"
+          fi
+          NEW_TAG="v${MAJOR}.${MINOR}.${PATCH}"
           echo "new_tag=$NEW_TAG" >> "$GITHUB_OUTPUT"
-          echo "Bumping $LATEST -> $NEW_TAG"
+          echo "Bumping $LATEST -> $NEW_TAG ($LEVEL)"
 
       - name: Create and push tag
         if: steps.version.outputs.new_tag

CLAUDE.md

@@ -136,7 +136,7 @@ Two separate Docker images:
 
 ### Authentication & Credential Flow
 
-- A single shared **API key** authenticates all workers to the UI (`CASHPILOT_API_KEY` env var).
+- **Per-worker fleet keys** (v1.0.0+): `CASHPILOT_API_KEY` is the shared **enrollment/bootstrap** key. On a worker's first heartbeat the UI issues it a unique key (stored encrypted on the UI, returned once, persisted on the worker under `/data/.worker_key`); thereafter each worker authenticates — in both directions — with its own key, and the shared key is rejected for an enrolled worker.
 - Credentials are Docker-native: UI sends full container spec (image, env vars, volumes, ports) to worker on deploy. Worker passes to Docker API. Docker stores env vars in container config. Restarts preserve env vars natively.
 
 ### Worker Environment Variables
@@ -172,7 +172,7 @@ Two separate Docker images:
 
 ### `release.yml` — Auto Release
 
-Triggers on push to `main` (paths: `app/`, `services/`, `Dockerfile*`, `requirements*.txt`). Auto-increments patch version, creates annotated git tag + GitHub Release. Skips if commit message contains `[skip ci]`.
+Triggers on push to `main` (paths: `app/`, `services/`, `Dockerfile*`, `requirements*.txt`). Chooses the version bump from conventional-commit markers since the last tag — **major** on a `BREAKING CHANGE` footer or `type!:` subject, **minor** on `feat:`, else **patch** — then creates an annotated git tag + GitHub Release. Skips if commit message contains `[skip ci]`.
 
 **CRITICAL: NEVER manually create tags or GitHub releases.** The workflow handles version bumping automatically. Manual tags cause `already_exists` conflicts and skip Docker builds.
 

README.md

@@ -157,7 +157,7 @@ cashpilot/
 |----------|---------|-------------|
 | `TZ` | `UTC` | Timezone for scheduling and display |
 | `CASHPILOT_SECRET_KEY` | *(auto-generated)* | Encryption key for stored credentials |
-| `CASHPILOT_API_KEY` | -- | Shared secret between UI and workers for API authentication |
+| `CASHPILOT_API_KEY` | -- | Enrollment/bootstrap key; each worker then gets its own key (per-worker fleet keys, v1.0.0+) |
 | `CASHPILOT_COLLECTION_INTERVAL` | `3600` | Seconds between earnings collection cycles |
 | `CASHPILOT_PORT` | `8080` | Web UI port inside the container |
 | `CASHPILOT_METRICS_ENABLED` | `false` | Set to `true` to expose Prometheus metrics at `/metrics` |

app/database.py

@@ -140,6 +140,8 @@ def decrypt_value(value: str) -> str:
     apps            TEXT    NOT NULL DEFAULT '[]',
     system_info     TEXT    NOT NULL DEFAULT '{}',
     last_heartbeat  TEXT,
+    api_key_enc     TEXT,
+    key_confirmed   INTEGER NOT NULL DEFAULT 0,
     registered_at   TEXT    NOT NULL DEFAULT (datetime('now'))
 );
 
@@ -323,6 +325,14 @@ async def init_db() -> None:
         elif "apps" not in cols:
             await db.execute("ALTER TABLE workers ADD COLUMN apps TEXT NOT NULL DEFAULT '[]'")
 
+        # Migrate workers table: add api_key_enc for per-worker fleet keys.
+        # (cols is the pre-rebuild snapshot; on a fresh DB the column comes from
+        # _SCHEMA so it is already present here and the ALTER is skipped.)
+        if "api_key_enc" not in cols:
+            await db.execute("ALTER TABLE workers ADD COLUMN api_key_enc TEXT")
+        if "key_confirmed" not in cols:
+            await db.execute("ALTER TABLE workers ADD COLUMN key_confirmed INTEGER NOT NULL DEFAULT 0")
+
         # 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()}
@@ -991,6 +1001,72 @@ async def delete_worker(worker_id: int) -> None:
         await db.close()
 
 
+# --- Per-worker fleet keys ---
+#
+# The UI must both VERIFY inbound heartbeats from a worker and, for the full
+# cutover, AUTHENTICATE outbound calls TO that worker — so it needs the key
+# itself, not just a one-way hash. Keys are therefore stored encrypted at rest
+# (Fernet, the same at-rest protection as service credentials) and decrypted on
+# demand for comparison and for outbound Authorization headers.
+
+
+async def set_worker_key(client_id: str, key: str) -> None:
+    """Store a worker's per-worker key (encrypted), unconfirmed until the worker
+    proves it holds the key by using it on a later heartbeat."""
+    db = await _get_db()
+    try:
+        cursor = await db.execute(
+            "UPDATE workers SET api_key_enc = ?, key_confirmed = 0 WHERE client_id = ?",
+            (encrypt_value(key), client_id),
+        )
+        await db.commit()
+        if not cursor.rowcount:
+            # The worker row must exist first (upsert runs before this); a missing
+            # row would silently drop the key and lock the worker out.
+            _logger.warning("set_worker_key: no worker row for client_id=%s", client_id)
+    finally:
+        await db.close()
+
+
+async def confirm_worker_key(client_id: str) -> None:
+    """Mark a worker's key confirmed — it has authenticated with its own key, so the
+    shared bootstrap key is refused from now on (the cutover finalizes)."""
+    db = await _get_db()
+    try:
+        await db.execute(
+            "UPDATE workers SET key_confirmed = 1 WHERE client_id = ?",
+            (client_id,),
+        )
+        await db.commit()
+    finally:
+        await db.close()
+
+
+async def get_worker_key(client_id: str) -> str | None:
+    """Return a worker's per-worker key (decrypted), or None if not yet enrolled."""
+    key, _ = await get_worker_key_state(client_id)
+    return key
+
+
+async def get_worker_key_state(client_id: str) -> tuple[str | None, bool]:
+    """Return (key, confirmed) for a worker: the decrypted per-worker key (or None
+    if unenrolled) and whether the worker has confirmed it by using it."""
+    db = await _get_db()
+    try:
+        cursor = await db.execute(
+            "SELECT api_key_enc, key_confirmed FROM workers WHERE client_id = ?",
+            (client_id,),
+        )
+        row = await cursor.fetchone()
+        if not row:
+            return None, False
+        enc = row["api_key_enc"]
+        key = decrypt_value(enc) if enc else None
+        return key, bool(row["key_confirmed"])
+    finally:
+        await db.close()
+
+
 # --- Health Events ---
 
 

app/main.py

@@ -14,6 +14,7 @@
 import logging
 import os
 import re
+import secrets
 import socket
 from collections import defaultdict
 from contextlib import asynccontextmanager
@@ -1006,9 +1007,15 @@ async def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[st
     if not worker["url"]:
         raise HTTPException(status_code=503, detail="Worker URL not known")
     url = await asyncio.to_thread(_validate_worker_url, worker["url"])
+    # Authenticate to the worker with ITS OWN key once enrolled; fall back to the
+    # shared bootstrap key only for workers that have not enrolled yet. Post-cutover
+    # an enrolled worker rejects the shared key, so the UI must present its own.
+    cid = worker.get("client_id") or ""
+    worker_key = await database.get_worker_key(cid) if cid else None
+    auth_key = worker_key or FLEET_API_KEY
     headers: dict[str, str] = {}
-    if FLEET_API_KEY:
-        headers["Authorization"] = f"Bearer {FLEET_API_KEY}"
+    if auth_key:
+        headers["Authorization"] = f"Bearer {auth_key}"
     return url, headers
 
 
@@ -1689,16 +1696,43 @@ async def api_admin_set_password(request: Request, user_id: int, body: AdminPass
 # ---------------------------------------------------------------------------
 
 
-def _verify_fleet_api_key(request: Request) -> None:
-    """Verify the shared fleet API key from a worker's request."""
+def _bearer_token(request: Request) -> str:
+    """Extract the bearer token from an Authorization header (empty if absent)."""
+    h = request.headers.get("Authorization", "")
+    return h[7:] if h.startswith("Bearer ") else ""
+
+
+async def _authenticate_worker_heartbeat(request: Request, cid: str) -> str:
+    """Authenticate a heartbeat and classify it. Returns one of:
+
+    - ``"enroll"``  — worker has no key yet and presented the shared key: mint one.
+    - ``"reissue"`` — worker has a key that is NOT yet confirmed and presented the
+      shared key: it likely lost the enrollment response, so re-deliver the SAME
+      key (until confirmed, the shared key still works for this one worker — a
+      bounded window that closes on the worker's first own-key heartbeat).
+    - ``"ok"``      — worker presented its own key: authenticated; confirm it so the
+      shared key is refused from now on (the cutover finalizes).
+
+    Raises 401 otherwise — notably, a confirmed worker presenting the shared key is
+    rejected, which is what stops shared-key holders from impersonating a worker.
+    """
     if not FLEET_API_KEY:
         raise HTTPException(
             status_code=503,
             detail="Fleet key not configured — set CASHPILOT_API_KEY or mount shared /fleet volume",
         )
-    auth_header = request.headers.get("Authorization", "")
-    if not hmac.compare_digest(auth_header.encode(), f"Bearer {FLEET_API_KEY}".encode()):
+    token = _bearer_token(request)
+    key, confirmed = await database.get_worker_key_state(cid) if cid else (None, False)
+    shared_ok = bool(token) and hmac.compare_digest(token.encode(), FLEET_API_KEY.encode())
+    if key is None:
+        if shared_ok:
+            return "enroll"
         raise HTTPException(status_code=401, detail="Invalid API key")
+    if token and hmac.compare_digest(token.encode(), key.encode()):
+        return "ok"
+    if not confirmed and shared_ok:
+        return "reissue"
+    raise HTTPException(status_code=401, detail="Invalid or missing per-worker key")
 
 
 class WorkerHeartbeat(BaseModel):
@@ -1713,9 +1747,11 @@ class WorkerHeartbeat(BaseModel):
 @app.post("/api/workers/heartbeat")
 async def api_worker_heartbeat(request: Request, body: WorkerHeartbeat) -> dict[str, Any]:
     """Receive a heartbeat from a worker. Registers or updates the worker."""
-    _verify_fleet_api_key(request)
     # Use client_id for identity; fall back to name for backward compat
     cid = body.client_id or body.name
+    if not cid:
+        raise HTTPException(status_code=400, detail="Worker name or client_id required")
+    state = await _authenticate_worker_heartbeat(request, cid)
     worker_id = await database.upsert_worker(
         client_id=cid,
         name=body.name,
@@ -1725,7 +1761,27 @@ async def api_worker_heartbeat(request: Request, body: WorkerHeartbeat) -> dict[
         system_info=json.dumps(body.system_info),
     )
     metrics.record_heartbeat(body.name)
-    return {"status": "ok", "worker_id": worker_id}
+    resp: dict[str, Any] = {"status": "ok", "worker_id": worker_id}
+    if state == "enroll":
+        # First contact: mint this worker's own key and hand it back once. Stored
+        # unconfirmed until the worker proves receipt by using it (see "ok").
+        new_key = secrets.token_urlsafe(32)
+        await database.set_worker_key(cid, new_key)
+        resp["worker_key"] = new_key
+        logger.info("Worker '%s' enrolled — per-worker key issued (awaiting confirmation)", cid)
+    elif state == "reissue":
+        # The worker still holds the shared key, so it never received/persisted its
+        # own key — re-deliver the same one so a dropped enrollment response can't
+        # lock it out.
+        existing = await database.get_worker_key(cid)
+        if existing:
+            resp["worker_key"] = existing
+        logger.info("Worker '%s' re-issued its per-worker key (not yet confirmed)", cid)
+    elif state == "ok":
+        # The worker authenticated with its own key: finalize the cutover so the
+        # shared key is refused from now on.
+        await database.confirm_worker_key(cid)
+    return resp
 
 
 @app.get("/api/workers")

app/worker_api.py

@@ -23,6 +23,7 @@
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime
 from html import escape as _esc
+from pathlib import Path
 from typing import Any
 
 import httpx
@@ -56,18 +57,58 @@
 _last_heartbeat: str = "never"
 _last_error: str = ""
 
+# Per-worker fleet key. On first contact the UI enrolls this worker and hands back
+# a key unique to us, which we persist here (in our own private /data, never the
+# shared /fleet volume) and use for all subsequent auth in both directions. Until
+# enrollment we authenticate with the shared bootstrap key.
+_WORKER_KEY_FILE = Path(os.getenv("CASHPILOT_DATA_DIR", "/data")) / ".worker_key"
+
+
+def _load_worker_key() -> str | None:
+    try:
+        if _WORKER_KEY_FILE.is_file():
+            return _WORKER_KEY_FILE.read_text().strip() or None
+    except OSError as exc:
+        logger.warning("Could not read per-worker key: %s", exc)
+    return None
+
+
+def _save_worker_key(key: str) -> None:
+    global _worker_key
+    _worker_key = key
+    try:
+        _WORKER_KEY_FILE.parent.mkdir(parents=True, exist_ok=True)
+        _WORKER_KEY_FILE.write_text(key)
+        _WORKER_KEY_FILE.chmod(0o600)
+    except OSError as exc:
+        logger.warning("Could not persist per-worker key: %s", exc)
+
+
+_worker_key: str | None = _load_worker_key()
+
+
+def _active_key() -> str:
+    """The key we authenticate with: our own once enrolled, else the shared key."""
+    return _worker_key or API_KEY
+
 
 # ---------------------------------------------------------------------------
 # Auth
 # ---------------------------------------------------------------------------
 
 
 def _verify_api_key(request: Request) -> None:
-    """Verify the shared API key from Authorization header."""
-    if not API_KEY:
+    """Verify an inbound UI->worker call.
+
+    Once enrolled we require OUR OWN per-worker key; the shared bootstrap key is
+    rejected (the cutover). Before enrollment we accept the shared key so the UI
+    can reach us to enroll in the first place.
+    """
+    expected = _active_key()
+    if not expected:
         raise HTTPException(status_code=503, detail="Fleet key not configured")
     auth = request.headers.get("Authorization", "")
-    if not hmac.compare_digest(auth.encode(), f"Bearer {API_KEY}".encode()):
+    if not hmac.compare_digest(auth.encode(), f"Bearer {expected}".encode()):
         raise HTTPException(status_code=401, detail="Invalid API key")
 
 
@@ -103,9 +144,16 @@ async def _send_heartbeat() -> None:
             resp = await client.post(
                 f"{UI_URL.rstrip('/')}/api/workers/heartbeat",
                 json=payload,
-                headers={"Authorization": f"Bearer {API_KEY}"},
+                headers={"Authorization": f"Bearer {_active_key()}"},
             )
             resp.raise_for_status()
+            # Enrollment: the UI returns our own per-worker key exactly once.
+            issued = None
+            with contextlib.suppress(Exception):
+                issued = resp.json().get("worker_key")
+            if issued and issued != _worker_key:
+                _save_worker_key(issued)
+                logger.info("Enrolled: received and persisted this worker's own fleet key")
             _ui_connected = True
             _last_heartbeat = datetime.now(UTC).strftime("%H:%M:%S UTC")
             _last_error = ""

docs/AUTOPILOT-WORKLOG.md

@@ -0,0 +1,35 @@
+# Autopilot Worklog — CashPilot per-worker-keys cutover
+
+Append-only. Newest at the bottom. Every "done" needs evidence (test/CI/commit).
+
+---
+
+### 2026-07-10 — kickoff
+- Branch `feat/per-worker-keys` created off `main` (after merging #90/#97/#98/#100).
+- Goal recorded in `docs/GOAL.md`: per-worker keys **full cutover** → `v1.0.0`, announced.
+- Established the plan (worker + UI + DB migration + release-workflow major bump + docs).
+- Next: DB layer first — `workers.api_key_hash` column + migration + helpers + tests.
+
+### 2026-07-10 — US-001 DONE (DB layer)
+- `workers.api_key_hash` column + idempotent migration; `hash_worker_key` (sha256),
+  `set/get_worker_key_hash`. Commit c22a5a3.
+- Evidence: 4 new TestWorkerKeys tests pass; full pytest **1085 passed**; ruff check+format clean.
+- Next: US-002 — UI heartbeat enrollment + reject shared key for enrolled workers.
+
+### 2026-07-10 — US-002/003 DONE (UI cutover, both directions)
+- US-002: heartbeat enrollment + reject shared key for enrolled workers (commit 088be34).
+- Storage upgraded hash->encrypted (api_key_enc) so the UI can also authenticate
+  outbound (commit 7328acf).
+- US-003: `_get_verified_worker_url` sends each worker's own key outbound (commit 7328acf+).
+- Evidence: full pytest **1088 passed**; ruff check+format clean.
+- Next: US-004 — worker side (obtain/persist/use its key; inbound verify accepts own key).
+
+### 2026-07-10 — US-003..007 DONE + architect review + PR opened
+- US-003 outbound (a3827ea), US-004 worker side (df3866f), US-005 release major-bump (0065304),
+  US-006 docs (f250799).
+- Architect (thorough) PASSED all 5 security criteria; fixed its MEDIUM availability
+  defect (lost-enrollment lockout) via key_confirmed re-delivery + low-sev items (becb5a1).
+- Deslop pass: no slop found (all helpers used, no dead code, purposeful comments).
+- US-007: **PR #101 opened** (feat(fleet)!: → v1.0.0), NOT merged (awaits Sergio).
+- Evidence: full pytest **1099 passed**; ruff check+format clean; branch pushed.
+- All 7 PRD stories passes:true. Waiting on #101 CI + CodeRabbit before cancel.