From db045770e886ca46e2c967ed3914913f2b8f18e7 Mon Sep 17 00:00:00 2001 From: stroblme Date: Tue, 6 Oct 2026 12:18:30 +0200 Subject: [PATCH] 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. --- backend/tests/flow/test_worker_agent.py | 77 +++++++++++++++++++++++++ docs/code/workers.md | 9 +++ worker/README.md | 4 ++ worker/fluksio_worker/agent.py | 56 +++++++++++++++++- worker/fluksio_worker/worker_main.py | 12 +++- 5 files changed, 156 insertions(+), 2 deletions(-) create mode 100644 backend/tests/flow/test_worker_agent.py diff --git a/backend/tests/flow/test_worker_agent.py b/backend/tests/flow/test_worker_agent.py new file mode 100644 index 0000000..e64290a --- /dev/null +++ b/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}) diff --git a/docs/code/workers.md b/docs/code/workers.md index b25822d..97f136c 100644 --- a/docs/code/workers.md +++ b/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) diff --git a/worker/README.md b/worker/README.md index 18f5c6c..c28ec42 100644 --- a/worker/README.md +++ b/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 diff --git a/worker/fluksio_worker/agent.py b/worker/fluksio_worker/agent.py index 93f0ec9..e9a12fc 100644 --- a/worker/fluksio_worker/agent.py +++ b/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") diff --git a/worker/fluksio_worker/worker_main.py b/worker/fluksio_worker/worker_main.py index 362f0ad..c09bde3 100644 --- a/worker/fluksio_worker/worker_main.py +++ b/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):