feat(compute): opt-in shared bearer token for the worker

.env.example

@@ -6,8 +6,9 @@ FUSIONLAB_DEVICE=
 
 # User-provided compute over SSH — see the README section "User-provided compute (SSH)" and
 # docs/COMPUTE.md. These override (or replace) the matching [compute] keys in fusionlab.toml at the
-# repo root; the env var wins. No key, password, or token belongs here: SSH auth comes from the
-# user's own config, agent, and keys.
+# repo root; the env var wins. No key, password, or SSH credential belongs here: SSH auth comes
+# from the user's own config, agent, and keys. The one app-level exception is the worker token
+# below, whose documented home is exactly here (same precedent as ANTHROPIC_API_KEY).
 # FUSIONLAB_COMPUTE_PROVIDER=local
 # FUSIONLAB_COMPUTE_HOST=gpu-lab
 # FUSIONLAB_SSH_HOST=
@@ -16,6 +17,12 @@ FUSIONLAB_DEVICE=
 # FUSIONLAB_SSH_PORT=8420
 # FUSIONLAB_SSH_REMOTE_PORT=8420
 
+# Optional shared token for the remote worker (#45): set the same value on the worker host's
+# environment and here, and every /v1 route on the worker demands `Authorization: Bearer` while
+# /health stays open for the tunnel probe. Unset (the default), the worker is loopback-first and
+# open, exactly as before the token existed.
+# FUSIONLAB_WORKER_TOKEN=
+
 # --- Optional in-app agent: all three are unnecessary for the offline demo -------------------
 # Master switch for the agent panel; optional, off unless exactly 1, true, or yes.
 FUSIONLAB_AGENT_ENABLED=

README.md

@@ -93,6 +93,11 @@ autostart = true            # start the worker over SSH when none is answering
 More than one GPU box? Define a `[compute.hosts.<name>]` profile per host (same keys) and switch
 with `FUSIONLAB_COMPUTE_HOST` — see [docs/COMPUTE.md](docs/COMPUTE.md).
 
+By default the worker binds loopback on the remote host and trusts the tunnel. If it might be
+reachable by more than your own tunnel, set `FUSIONLAB_WORKER_TOKEN` to the same value on both ends
+(`.env` here, the worker's environment there) and every `/v1` route on the worker demands a Bearer
+token — same file, no new services; details in [docs/COMPUTE.md](docs/COMPUTE.md).
+
 Full guide — prerequisites, what the chip and banner mean, and a first-run checklist against a real
 host: [docs/COMPUTE.md](docs/COMPUTE.md).
 

docs/COMPUTE.md

@@ -9,7 +9,8 @@ streams results back.
 
 The compute side of the connection is your machine, your account, your SSH setup. FusionLab adds no
 credentials of its own: it uses your keys, agent, and `~/.ssh/config` exactly as they are, and
-neither `fusionlab.toml` nor `.env` ever holds a key, password, or token.
+neither `fusionlab.toml` nor `.env` ever holds a key, password, or token. (The one deliberate
+exception is the optional worker token below — env-only, on both ends; never in `fusionlab.toml`.)
 
 ## Quick start
 
@@ -129,14 +130,24 @@ checklist is the real-host path that remains.
   remote jobs only.
 - **The worker executes a fixed registry of named tasks** — field-line tracing, the two
   surrogates, the model-metrics readout, benchmarks. It is not a shell: schema-checked inputs, no
-  command execution, payload-size caps. It binds loopback only, reachable through the tunnel.
+  command execution, payload-size caps. It binds loopback only, reachable through the tunnel — and
+  if you expose it further, an optional shared token covers the whole `/v1` surface (below).
 - **Shot data** is read by the worker through the same FAIR-MAST path as everywhere else, from its
   own clone (which ships the same small showcase cache; other shots fetch on demand). Compact
   inputs — EFIT equilibria, signal arrays — ride inside the job itself. So the compute host needs
   ordinary internet, not a copy of the archive.
 - **Version drift** between your laptop's clone and the worker's is surfaced, not hidden: the
   handshake reports the worker's version and the chip warns on mismatch. The fix is the usual one
   on the remote side — `git pull` and `uv sync`.
-- **Not in this version**: SLURM/PBS submission, multi-user queues or auth, training jobs, Windows
-  remotes, and anything that opens a port on your network. If you cannot install software on the
-  compute target, this design does not fit it — that is its one real prerequisite.
+- **An optional shared token locks the worker's `/v1` surface** (issue #45): if the worker might be
+  reachable by more than your own tunnel, set `FUSIONLAB_WORKER_TOKEN` to the same value on both
+  ends (the laptop's `.env` and the worker host's environment) and every `/v1` route on the worker
+  demands `Authorization: Bearer` — constant-time compare, 401 before any request is parsed or
+  queued — while `/health` stays open so the tunnel probe and autostart keep working. Unset (the
+  default), the worker is exactly as open as before. When the two ends disagree, the fallback
+  banner names the token mismatch instead of a generic "unreachable", and the handshake advertises
+  the mode (`"auth": "token" | "open"`) so the two sides can see the disagreement coming.
+- **Not in this version**: SLURM/PBS submission, multi-user queues or per-user accounts (the token
+  is one shared secret, not an account system), training jobs, Windows remotes, and anything that
+  opens a port on your network. If you cannot install software on the compute target, this design
+  does not fit it — that is its one real prerequisite.

fusionlab.toml.example

@@ -1,7 +1,9 @@
 # FusionLab compute configuration (optional). Copy to fusionlab.toml at the repo root, beside .env.
 # Every value can also be set with a FUSIONLAB_* environment variable (see .env.example); the env
 # var wins over this file. Neither place ever holds a key, password, or token — the connection uses
-# the user's own SSH config, agent, and keys (design §8).
+# the user's own SSH config, agent, and keys (design §8). The worker's shared token is env-only by
+# design (FUSIONLAB_WORKER_TOKEN in .env, both ends); a credential-shaped key in this file is a
+# loud config error, never a silently-carried option.
 
 [compute]
 # "local" (default): compute runs in this process — zero config, works offline.

fusionlab/compute/__init__.py

@@ -36,6 +36,10 @@
     "remote_port": "FUSIONLAB_SSH_REMOTE_PORT",
 }
 
+# The worker's shared token (#45) is env-only: FUSIONLAB_WORKER_TOKEN is deliberately absent from the
+# override map above, and a credential-shaped key in the toml is a loud error, never a carried option.
+_CREDENTIAL_KEYS = ("token", "worker_token", "password", "secret", "api_key")
+
 
 def _ssh_table(cfg: dict[str, Any]) -> dict[str, Any]:
     """The base ssh table for the selected host — raw [compute.ssh]-shaped keys, no env merging yet.
@@ -70,7 +74,8 @@ def ssh_options() -> dict[str, Any]:
     from fusionlab.toml, merged with FUSIONLAB_SSH_* env overrides (env wins).
 
     Returns kwargs ready for SshProvider(**...). Never contains credentials — the connection uses the
-    user's own SSH agent, keys, and ~/.ssh/config (design §8).
+    user's own SSH agent, keys, and ~/.ssh/config (design §8), and a credential-shaped key in the
+    toml is a loud ValueError, never a silently-carried option (the worker token is env-only, #45).
     """
     opts: dict[str, Any] = {}
     if _TOML_PATH.exists():
@@ -88,6 +93,11 @@ def ssh_options() -> dict[str, Any]:
             opts[key] = int(raw)
         else:
             opts[key] = raw
+    leaked = sorted(set(opts) & set(_CREDENTIAL_KEYS))
+    if leaked:
+        raise ValueError(f"fusionlab.toml holds credential key(s) {leaked} — config never holds a key, "
+                         "password, or token (see fusionlab.toml.example): the worker token is the "
+                         "FUSIONLAB_WORKER_TOKEN environment variable")
     return opts
 
 

fusionlab/compute/ssh.py

@@ -36,6 +36,7 @@
 import importlib.metadata
 import json
 import logging
+import os
 import threading
 import time
 from collections.abc import AsyncIterator
@@ -58,6 +59,14 @@
 _TERMINAL_EVENTS = ("completed", "error", "cancelled")
 
 
+def _worker_auth_headers() -> dict[str, str]:
+    """Authorization headers for every worker HTTP call, from FUSIONLAB_WORKER_TOKEN — empty when no
+    token is configured (opt-in by absence: a tokenless backend against an open worker is today's
+    behavior, byte for byte; a token the worker doesn't expect is a plain header it ignores)."""
+    token = os.environ.get("FUSIONLAB_WORKER_TOKEN", "")
+    return {"Authorization": f"Bearer {token}"} if token else {}
+
+
 class RemoteComputeError(RuntimeError):
     """Any SshProvider failure the caller should handle by falling back to LocalProvider.
 
@@ -71,6 +80,20 @@ def __init__(self, phase: str, message: str):
         self.message = message
 
 
+def _raise_for_auth(r: httpx.Response) -> None:
+    """Map a worker 401 to the one actionable auth error, at every HTTP call site.
+
+    A 401 means the two ends disagree about FUSIONLAB_WORKER_TOKEN — say exactly that, so the
+    fallback banner reads "token differs" instead of a generic unreachable (run_task records this
+    message and GET /compute renders it verbatim)."""
+    if r.status_code == 401:
+        raise RemoteComputeError(
+            "auth",
+            "worker rejected the token — FUSIONLAB_WORKER_TOKEN differs between this machine and "
+            "the worker host (or is set on one side only)",
+        )
+
+
 class WorkerSession(Protocol):
     """What SshProvider needs from a transport: a loopback HTTP base_url that answers after ensure()."""
 
@@ -174,7 +197,7 @@ async def _wait_worker(self) -> None:
         started = False
         while True:
             try:
-                async with httpx.AsyncClient() as client:
+                async with httpx.AsyncClient(headers=_worker_auth_headers()) as client:
                     r = await client.get(f"{self.base_url}/health", timeout=2.0)
                 if r.status_code == 200 and r.json().get("ok"):
                     return
@@ -255,7 +278,8 @@ def __init__(self, session: WorkerSession | None = None, *,
             self._session = _SshSession(host, **session_kw)
         self._result_timeout_s = result_timeout_s
         self._remote: dict[str, Any] | None = None   # the worker's last handshake
-        self._http = httpx.Client(base_url=self._session.base_url, timeout=30.0)
+        self._http = httpx.Client(base_url=self._session.base_url, timeout=30.0,
+                                  headers=_worker_auth_headers())
 
     # -- ComputeProvider protocol
     def capabilities(self) -> dict:
@@ -281,6 +305,7 @@ def submit(self, task: str, inputs: dict) -> JobHandle:
             payload = {"task": task, "inputs": frames.encode_inputs(inputs),
                        "artifact_refs": self._ensure_artifacts(task)}
             r = self._http.post("/v1/jobs", json=payload)
+            _raise_for_auth(r)
             if r.status_code == 409:
                 raise RemoteComputeError("submit", r.json().get("detail", "worker busy"))
             if r.status_code == 422:
@@ -301,10 +326,12 @@ async def stream(self, job: JobHandle) -> AsyncIterator[dict]:
         """
         sse_timeouts = httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0)
         try:
-            async with httpx.AsyncClient(base_url=self._session.base_url, timeout=sse_timeouts) as client, \
+            async with httpx.AsyncClient(base_url=self._session.base_url, timeout=sse_timeouts,
+                                         headers=_worker_auth_headers()) as client, \
                     client.stream("GET", f"/v1/jobs/{job.job_id}/events") as resp:
                 if resp.status_code == 404:
                     raise RemoteComputeError("job", f"unknown job {job.job_id} (worker restarted?)")
+                _raise_for_auth(resp)
                 resp.raise_for_status()
                 event: dict = {}
                 async for line in resp.aiter_lines():
@@ -329,6 +356,7 @@ def cancel(self, job: JobHandle) -> None:
             r = self._http.post(f"/v1/jobs/{job.job_id}/cancel")
         except httpx.HTTPError as e:
             raise RemoteComputeError("job", f"cancel failed: {type(e).__name__}: {e}") from None
+        _raise_for_auth(r)
         if r.status_code == 409:
             return
         if r.status_code >= 400:
@@ -343,6 +371,7 @@ def result(self, job: JobHandle) -> Any:
                 r = self._http.get(f"/v1/jobs/{job.job_id}")
             except httpx.HTTPError as e:
                 raise RemoteComputeError("collect", f"{type(e).__name__}: {e}") from None
+            _raise_for_auth(r)
             if r.status_code == 404:
                 raise RemoteComputeError("collect", f"unknown job {job.job_id} (worker restarted?)")
             r.raise_for_status()
@@ -361,6 +390,7 @@ def result(self, job: JobHandle) -> Any:
             rr = self._http.get(f"/v1/jobs/{job.job_id}/result")
         except httpx.HTTPError as e:
             raise RemoteComputeError("collect", f"{type(e).__name__}: {e}") from None
+        _raise_for_auth(rr)
         if rr.status_code != 200:
             raise RemoteComputeError("collect", f"result fetch failed: HTTP {rr.status_code}")
         try:
@@ -380,6 +410,7 @@ def health(self) -> None:
         RemoteComputeError, the one typed signal GET /compute renders as "unreachable"."""
         try:
             r = self._http.get("/health", timeout=2.0)
+            _raise_for_auth(r)   # /health is open today; if a future worker ever gates it, say "auth"
             r.raise_for_status()
         except httpx.HTTPError as e:
             try:
@@ -394,6 +425,7 @@ def _handshake(self) -> dict:
             try:
                 self._session.ensure()
                 r = self._http.get("/v1/capabilities")
+                _raise_for_auth(r)
                 r.raise_for_status()
                 self._remote = r.json()
             except httpx.HTTPError as e:
@@ -422,6 +454,7 @@ def _ensure_artifacts(self, task: str) -> list[dict[str, str]]:
     def _upload(self, path: Path, digest: str) -> None:
         try:
             r = self._http.post(f"/v1/artifacts/{digest}", content=path.read_bytes())
+            _raise_for_auth(r)
             if r.status_code != 201:
                 raise RemoteComputeError("submit", f"weight upload failed: HTTP {r.status_code} {r.text[:200]}")
             logger.info("uploaded models/%s (%d bytes) to the worker cache — one-time per hash",

fusionlab/worker.py

@@ -9,13 +9,20 @@
 an artifact cache for trained weights: no shell, no eval, no filesystem passthrough. Worst case if
 an exposed worker were found by a stranger is compute theft, not host compromise (design §8).
 
+For a worker that is not loopback-only, an optional shared token locks the whole /v1 surface: set
+FUSIONLAB_WORKER_TOKEN to the same value on both ends (the worker's environment and the backend's)
+and every /v1 route demands `Authorization: Bearer` — constant-time compare, 401 before any body
+parsing or queue interaction — while /health stays open, because a token-gated probe would be
+misread as "worker absent" and trigger a doomed autostart. Unset (the default) nothing changes.
+
     POST /v1/jobs               {task, inputs, artifact_refs} -> {job_id, state}; 409 when busy
     GET  /v1/jobs/{id}/events   SSE: queued / running / progress / completed / error / cancelled
+    GET  /v1/jobs/{id}          state without the payload — the poll target while a job runs
     GET  /v1/jobs/{id}/result   application/x-fusionlab-frames (fusionlab.frames), consumed once
     POST /v1/jobs/{id}/cancel   cooperative: immediate while queued; benchmark tasks check between sections
     POST /v1/artifacts/{sha256} upload one trained weight into the cache (hash-verified, capped)
-    GET  /v1/capabilities       handshake: worker/torch/warp versions, CUDA, tasks, weight hashes
-    GET  /health                {ok: true} — the tunnel/autostart probe
+    GET  /v1/capabilities       handshake: worker/torch/warp versions, CUDA, tasks, weight hashes, auth mode
+    GET  /health                {ok: true} — the tunnel/autostart probe, always open
 
 One job at a time (a single-user GPU box is honest about itself); a second submission gets a clean
 409 instead of a hidden queue. Jobs live in memory only — a worker restart forgets them, and the
@@ -28,14 +35,15 @@
 import asyncio
 import contextlib
 import hashlib
+import hmac
 import importlib.metadata
 import json
 import os
 import platform
 import re
 import threading
 import time
-from collections.abc import AsyncIterator, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Literal
@@ -369,6 +377,19 @@ def execute() -> bytes:
         _JOBS.append(job, {"event": "error", "error": str(e), "error_type": type(e).__name__})
 
 
+# ---------------------------------------------------------------- auth (opt-in shared token)
+def worker_token() -> str:
+    """The shared bearer token, read per request (env-only by design — config files never hold one;
+    tests monkeypatch the environment). Empty means the worker is open: every route behaves exactly
+    as before the token existed."""
+    return os.environ.get("FUSIONLAB_WORKER_TOKEN", "")
+
+
+def auth_mode() -> str:
+    """What the handshake advertises: 'token' when a shared token is set, else 'open'."""
+    return "token" if worker_token() else "open"
+
+
 # ---------------------------------------------------------------- handshake
 def worker_version() -> str:
     try:
@@ -383,7 +404,8 @@ def worker_version() -> str:
 
 def handshake() -> dict[str, Any]:
     """Version/device identity (computed once — a running box does not change its GPU), then the
-    per-call parts: tasks, shipped weight hashes, and what the artifact cache already holds."""
+    per-call parts: tasks, shipped weight hashes, what the artifact cache already holds, and the
+    auth mode."""
     global _HANDSHAKE
     with _HANDSHAKE_LOCK:
         if _HANDSHAKE is None:
@@ -397,13 +419,34 @@ def handshake() -> dict[str, Any]:
             _HANDSHAKE = {"worker_protocol_version": WORKER_PROTOCOL_VERSION, "worker_version": worker_version(),
                           "torch": torch.__version__, "warp": warp_version, "cuda_available": cuda,
                           "device_name": torch.cuda.get_device_name(0) if cuda else (platform.processor() or "CPU")}
-    return {**_HANDSHAKE, "tasks": sorted(TASKS), "models": shipped_models(), "artifacts": cached_artifacts()}
+    return {**_HANDSHAKE, "tasks": sorted(TASKS), "models": shipped_models(),
+            "artifacts": cached_artifacts(), "auth": auth_mode()}
 
 
 # ---------------------------------------------------------------- the app
 app = FastAPI(title="fusionlab worker", version=worker_version(), docs_url=None, redoc_url=None)
 
 
+@app.middleware("http")
+async def require_bearer_token(request: Request,
+                               call_next: Callable[[Request], Awaitable[Response]]) -> Response:
+    """Opt-in shared-token gate (issue #45): with FUSIONLAB_WORKER_TOKEN set, every /v1 route demands
+    `Authorization: Bearer <token>` (constant-time compare) and rejects anything else with 401 +
+    WWW-Authenticate BEFORE any body parsing or queue interaction. /health stays outside the check:
+    a token-gated probe would be misread as "worker absent" and trigger a doomed autostart."""
+    token = worker_token()
+    if token and request.url.path.startswith("/v1/"):
+        supplied = request.headers.get("authorization", "")
+        if not hmac.compare_digest(supplied, f"Bearer {token}"):
+            return JSONResponse(
+                status_code=401,
+                content={"detail": "missing or mismatched bearer token — set FUSIONLAB_WORKER_TOKEN, "
+                                   "the same value on both ends of the tunnel"},
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+    return await call_next(request)
+
+
 @app.get("/health")
 async def health() -> dict:
     return {"ok": True}