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.

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):