Harden the remote worker: TLS off loopback, full digests, filtered env

fluksio-worker now refuses ws:// and http:// engine and artifact URLs unless
they are loopback or --insecure is given; the engine still accepts both.
An artifact reference must be sha256: plus 64 hex digits before it is joined
into a path, and node code gets the host environment through the engine's
credential denylist instead of all of it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ

backend/tests/flow/test_worker_agent.py

@@ -0,0 +1,77 @@
+"""The remote worker's own guards: where its token goes, and what node code gets."""
+
+import argparse
+
+import pytest
+from fluksio_worker import agent
+from fluksio_worker.worker_main import _Reporter
+
+
+@pytest.mark.parametrize(
+    ("url", "trusted"),
+    [
+        ("wss://engine.example.com/api/v1/workers/attach", True),
+        ("https://engine.example.com/api/v1/artifacts", True),
+        ("ws://127.0.0.1:8000/api/v1/workers/attach", True),
+        ("ws://127.1.2.3:8000/api/v1/workers/attach", True),
+        ("ws://[::1]:8000/api/v1/workers/attach", True),
+        ("ws://localhost:8000/api/v1/workers/attach", True),
+        ("ws://192.0.2.10:8000/api/v1/workers/attach", False),
+        ("ws://engine.local:8000/api/v1/workers/attach", False),
+        ("http://192.0.2.10:8000/api/v1/artifacts", False),
+    ],
+)
+def test_only_tls_or_this_machine_is_handed_the_token(url: str, trusted: bool) -> None:
+    assert agent._encrypted_or_local(url) is trusted
+
+
+def test_a_cleartext_engine_elsewhere_needs_insecure(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    url = "ws://192.0.2.10:8000/api/v1/workers/attach"
+    with pytest.raises(SystemExit):
+        agent.main(["--url", url, "--token", "t"])
+    assert "--insecure" in capsys.readouterr().err
+
+    async def attached(self: agent.Agent) -> None:
+        return None
+
+    monkeypatch.setattr(agent.Agent, "serve_forever", attached)
+    assert agent.main(["--url", url, "--token", "t", "--insecure"]) == 0
+
+
+def test_node_code_does_not_get_the_hosts_credentials(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("FLUKSIO_WORKER_TOKEN", "worker-token")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "not-for-nodes")
+    monkeypatch.setenv("CUDA_VISIBLE_DEVICES", "2")
+    args = argparse.Namespace(
+        url="wss://engine.example.com/api/v1/workers/attach",
+        artifact_url="",
+        token="worker-token",
+    )
+
+    env = agent.Agent(args).env
+
+    assert "FLUKSIO_WORKER_TOKEN" not in env
+    assert "AWS_SECRET_ACCESS_KEY" not in env
+    assert env["CUDA_VISIBLE_DEVICES"] == "2"
+    assert "PATH" in env
+    # Put back on purpose: the artifact store is reached with it.
+    assert env["FLUKSIO_ARTIFACT_URL"] == "https://engine.example.com/api/v1/artifacts"
+
+
+@pytest.mark.parametrize(
+    "digest",
+    [
+        "sha256:../../../etc/passwd",
+        "sha256:" + "a" * 63,
+        "sha256:" + "A" * 64,
+        "md5:" + "a" * 64,
+    ],
+)
+def test_only_a_whole_digest_is_opened(digest: str) -> None:
+    """The digest is joined into a path, so anything else could read any file."""
+    with pytest.raises(ValueError, match="not an artifact reference"):
+        _Reporter("fluksio").load_artifact({"digest": digest})

docs/code/workers.md

@@ -39,6 +39,15 @@ driver in order to run a training step. An engine host already has it, and
 | `--gpus` | what the job says, else counted | GPUs to advertise |
 | `--ram-mb` | what the job or the machine has | memory to advertise, in MB |
 | `--max-idle` | never | stop after this many seconds with nothing running |
+| `--insecure` | off | allow `ws://` and `http://` to an engine on another machine |
+
+The worker refuses a plain `ws://` or `http://` address unless the engine is on
+this machine (`localhost`, `127.0.0.0/8`, `::1`): its token and every secret a
+node is handed would cross the network in clear. `--insecure` is for a network
+you trust, such as an engine on the same LAN without TLS. Node code gets this
+host's environment less anything named like a credential (`*TOKEN*`,
+`*SECRET*`, `*PASSWORD*`), the same as on the engine; hand a node a credential
+as a [secret](../interface/operations.md#secrets) instead.
 
 `--python` is the important one. It is how this machine keeps its own wheels
 (the CUDA build, the vendor SDK, the thing that will not install anywhere else)

worker/README.md

@@ -18,6 +18,10 @@ fluksio-worker \
 its own wheels without the engine ever installing them. Mint a token from the
 engine with `POST /api/v1/workers/tokens`.
 
+A plain `ws://` address is refused unless the engine is on this machine, since
+the token and node secrets would cross the network in clear; `--insecure`
+allows it on a network you trust.
+
 Linux and macOS.
 
 ## License

worker/fluksio_worker/agent.py

@@ -29,6 +29,7 @@ import argparse
 import asyncio
 import contextlib
 import glob
+import ipaddress
 import json
 import logging
 import os
@@ -38,6 +39,7 @@ import sys
 import time
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlsplit
 
 try:
     import websockets
@@ -54,6 +56,9 @@ PROTOCOL = 2
 HEARTBEAT_S = 10.0
 #: Reconnection backs off to this and no further.
 MAX_BACKOFF_S = 30.0
+#: Environment node code is not given; the engine's lists, see `_node_env`.
+ENV_DENY_PREFIXES = ("DATABASE_URL", "FIRST_SUPERUSER", "SENTRY_DSN")
+ENV_DENY_WORDS = ("PASSWORD", "SECRET", "TOKEN")
 
 
 class Refused(RuntimeError):
@@ -125,7 +130,7 @@ class Agent:
 
     def __init__(self, args: argparse.Namespace) -> None:
         self.args = args
-        self.env = dict(os.environ)
+        self.env = _node_env()
         self.env["FLUKSIO_ARTIFACT_URL"] = args.artifact_url or _artifacts_from(
             args.url
         )
@@ -305,6 +310,41 @@ def _report_failure(task: asyncio.Task[Any]) -> None:
         log.exception("call failed", exc_info=task.exception())
 
 
+def _node_env() -> dict[str, str]:
+    """This host's environment with its credentials taken out, for node code.
+
+    The engine's own denylist (``fluksio.flow.workers.worker_env``), copied
+    because nothing of the engine is imported here: ``PATH``, ``HOME`` and
+    whatever a venv needs still arrive, a ``*_TOKEN`` or ``*_PASSWORD`` does
+    not — this worker's own token included. Not a sandbox: node code runs as
+    the same user on the same machine.
+    """
+    return {
+        key: value
+        for key, value in os.environ.items()
+        if not key.startswith(ENV_DENY_PREFIXES)
+        and not any(word in key.upper() for word in ENV_DENY_WORDS)
+    }
+
+
+def _encrypted_or_local(url: str) -> bool:
+    """Whether the token and node secrets sent to *url* stay off the wire.
+
+    TLS, or this machine. An engine across a network on plain ``ws://`` hands
+    both to anything on the path.
+    """
+    parts = urlsplit(url)
+    if parts.scheme in ("wss", "https"):
+        return True
+    host = parts.hostname or ""
+    if host == "localhost":
+        return True
+    try:
+        return ipaddress.ip_address(host).is_loopback
+    except ValueError:
+        return False
+
+
 def _artifacts_from(url: str) -> str:
     """The artifact endpoint beside the socket, so one URL configures both."""
     base = url.replace("wss://", "https://").replace("ws://", "http://")
@@ -421,6 +461,14 @@ def main(argv: list[str] | None = None) -> int:
     )
     parser.add_argument("--parallel", type=int, default=1)
     parser.add_argument("--artifact-url", default="")
+    parser.add_argument(
+        "--insecure",
+        action="store_true",
+        help=(
+            "allow ws:// and http:// to an engine off this machine — the token "
+            "and every secret a node is handed then cross the network in clear"
+        ),
+    )
     parser.add_argument(
         "--cpus",
         type=int,
@@ -452,6 +500,12 @@ def main(argv: list[str] | None = None) -> int:
 
     if not args.token:
         parser.error("a token is required (--token or FLUKSIO_WORKER_TOKEN)")
+    for url in (args.url, args.artifact_url):
+        if url and not args.insecure and not _encrypted_or_local(url):
+            parser.error(
+                f"{url} is not encrypted: use wss:// (https:// for --artifact-url), "
+                "or --insecure on a network you trust"
+            )
     args.labels = [part.strip() for part in args.labels.split(",") if part.strip()]
     if not WORKER_MAIN.exists():
         parser.error(f"{WORKER_MAIN} is missing — copy it beside this file")

worker/fluksio_worker/worker_main.py

@@ -144,7 +144,7 @@ class _Reporter(ModuleType):
     def load_artifact(self, ref: dict[str, Any]) -> str:
         """Fetch an artifact and hand back a local path to read it from."""
         digest = str((ref or {}).get("digest") or "")
-        if not digest.startswith("sha256:"):
+        if not _valid_digest(digest):
             raise ValueError("not an artifact reference")
         return _fetch(digest)
 
@@ -318,6 +318,16 @@ def _put_over_http(
         pass
 
 
+def _valid_digest(digest: str) -> bool:
+    """The engine's own check: the digest becomes part of a path below."""
+    body = digest.removeprefix("sha256:")
+    return (
+        digest.startswith("sha256:")
+        and len(body) == 64
+        and all(c in "0123456789abcdef" for c in body)
+    )
+
+
 def _fetch(digest: str) -> str:
     """A local path holding this artifact's bytes, downloading it if needed."""
     for variable in (ARTIFACT_DIR_ENV, ARTIFACT_VOLATILE_DIR_ENV):