diff --git a/.secrets.baseline b/.secrets.baseline index 1d6cae90..2e37bb96 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -229,21 +229,21 @@ "filename": "fluid_build/build_runners/_acquisition_common.py", "hashed_secret": "3ea6cf17f95e272933a3b927317b7210d1bac5d5", "is_verified": false, - "line_number": 65 + "line_number": 70 }, { "type": "Secret Keyword", "filename": "fluid_build/build_runners/_acquisition_common.py", "hashed_secret": "fef341f85d87439e7d91a2d465b9871ef66b5e98", "is_verified": false, - "line_number": 515 + "line_number": 520 }, { "type": "Secret Keyword", "filename": "fluid_build/build_runners/_acquisition_common.py", "hashed_secret": "0c87bed4927509bf93be6f005081eddfdaf53688", "is_verified": false, - "line_number": 520 + "line_number": 525 } ], "fluid_build/cli/_forge_data_model_jdbc.py": [ @@ -601,49 +601,49 @@ "filename": "tests/build_runners/test_acquisition_common.py", "hashed_secret": "0c87bed4927509bf93be6f005081eddfdaf53688", "is_verified": false, - "line_number": 129 + "line_number": 144 }, { "type": "Secret Keyword", "filename": "tests/build_runners/test_acquisition_common.py", "hashed_secret": "e493f561d90c6638c1f51c5a8a069c3b129b79ed", "is_verified": false, - "line_number": 136 + "line_number": 151 }, { "type": "Secret Keyword", "filename": "tests/build_runners/test_acquisition_common.py", "hashed_secret": "2a31c218d1fac3715608a988676ef7f1f3ef1c6d", "is_verified": false, - "line_number": 147 + "line_number": 162 }, { "type": "Secret Keyword", "filename": "tests/build_runners/test_acquisition_common.py", "hashed_secret": "7a12adcc1f84a824c761e5cdabbdf8e5d3687136", "is_verified": false, - "line_number": 161 + "line_number": 176 }, { "type": "Secret Keyword", "filename": "tests/build_runners/test_acquisition_common.py", "hashed_secret": "c55ffb28589b12e4fe0044911a353dd7e012b703", "is_verified": false, - "line_number": 171 + "line_number": 186 }, { "type": "Secret Keyword", "filename": "tests/build_runners/test_acquisition_common.py", "hashed_secret": "fef341f85d87439e7d91a2d465b9871ef66b5e98", "is_verified": false, - "line_number": 281 + "line_number": 296 }, { "type": "Secret Keyword", "filename": "tests/build_runners/test_acquisition_common.py", "hashed_secret": "be3bbce74ff0f4c3dcd5e4619c15b1e6c16fcf46", "is_verified": false, - "line_number": 516 + "line_number": 531 } ], "tests/build_runners/test_airbyte_host_override.py": [ @@ -2154,5 +2154,5 @@ } ] }, - "generated_at": "2026-10-22T06:15:34Z" + "generated_at": "2026-10-26T19:01:07Z" } diff --git a/CHANGELOG.md b/CHANGELOG.md index 54aabcf7..08fd94b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,51 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 declared DSN still carries its own credentials, so a legitimate source is unaffected and no contract change is needed. +- **A contract could send any environment variable to a host of its own choosing.** + `connection.secretRef: env://NAME` let the contract name both halves — the variable to read and + the host the resolved value is sent to as a password — and nothing checked either. A run whose + environment holds secrets (CI, an operator's shell) could therefore have any variable + exfiltrated. All six acquisition runners (airbyte, debezium, dlt, duckdb, kafka_connect, + meltano) go through one helper, so all six were affected. + `env://` now resolves only when **every** destination the connection block carries is loopback, + or when the operator has listed the variable in `FLUID_ACQUISITION_SECRET_REF_ALLOWLIST` as + `NAME` (any host) or `NAME@host`. Every destination is checked, not just the first field that + parses, so a decoy `host: localhost` beside a remote `connection_string` does not clear the + send. The check runs **before** the reference is resolved, so a refused send never reads the + secret, and the refusal names the variable and host but never the value. `vault://`, `aws://`, + `gcp://`, `azure://` and `file://` are not policed: those name a secret the operator provisioned + and granted this process access to, which is already an operator decision. A `secretRef` whose + target field already holds an inline credential is not policed either — the reference is never + read and nothing is sent. + The allowlist is an **operator** setting: a value set by one of the project's own dotenv files is + ignored, because `fluid apply` hydrates the environment from `.env`, `.env.` and + `FLUID_SECRETS_FILE`, and those files are committed beside the contract. +- **`TESTCONTAINERS_HOST_OVERRIDE` is now an operator-only setting**, as + `FLUID_RUNNER_HOST_OVERRIDE` already was. Both rewrite a contract-declared loopback host to a + reachable one *after* the policy above has cleared the send as local-only, so a project dotenv + file able to set either could have redirected a credential allowed only because it was staying + on this machine. +- Resolved `secretRef` values are registered with the exact-value redactor (#884), so a literal a + client library echoes in its own error is masked by value rather than by pattern. + + +- A contract repository could hand the whole host to its own SQL through DuckDB's spill + directory. `secure_duckdb_connect` never set `temp_directory`, and DuckDB resolves its + temp directory into `allowed_directories` like any other grant; the default is relative + (`.tmp` in the working directory, or `.tmp` beside a file database), so a repository + shipping `.tmp -> /` got `allowed_directories = ['/']` on a connection that granted + nothing, and contract SQL could read any file the process could. Reproduced on duckdb + 1.5.1, 1.5.6 and 2.0.0, for an in-memory and a file database, through the same helper the + MCP output port's `query_sql` runs on. The engine now creates and owns that directory + (mode 0700, under `$TMPDIR`, removed with the connection), reads the grants back under + the lock and refuses the connection if DuckDB took a directory the call site did not + name, and refuses a caller-supplied `temp_directory` rather than silently overriding it. + Spilling is unaffected. No contract, flag or configuration change is needed. +- Opening one DuckDB database file from two threads at once is now serialised. DuckDB + shares one instance per file per process and the sandbox's settings live on the instance, + so the two opens could cross-configure: a connection could be returned running under the + other thread's allowlist. The second opener now reliably gets the existing + "already open in this process" refusal. ### Fixed @@ -74,27 +119,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 with an invalid type stay invalid — the fix is emit-time — but those already failed validation, so nothing working changes. -### Security +### Known limitations -- A contract repository could hand the whole host to its own SQL through DuckDB's spill - directory. `secure_duckdb_connect` never set `temp_directory`, and DuckDB resolves its - temp directory into `allowed_directories` like any other grant; the default is relative - (`.tmp` in the working directory, or `.tmp` beside a file database), so a repository - shipping `.tmp -> /` got `allowed_directories = ['/']` on a connection that granted - nothing, and contract SQL could read any file the process could. Reproduced on duckdb - 1.5.1, 1.5.6 and 2.0.0, for an in-memory and a file database, through the same helper the - MCP output port's `query_sql` runs on. The engine now creates and owns that directory - (mode 0700, under `$TMPDIR`, removed with the connection), reads the grants back under - the lock and refuses the connection if DuckDB took a directory the call site did not - name, and refuses a caller-supplied `temp_directory` rather than silently overriding it. - Spilling is unaffected. No contract, flag or configuration change is needed. -- Opening one DuckDB database file from two threads at once is now serialised. DuckDB - shares one instance per file per process and the sandbox's settings live on the instance, - so the two opens could cross-configure: a connection could be returned running under the - other thread's allowlist. The second opener now reliably gets the existing - "already open in this process" refusal. +- **The loopback exemption covers only the engines where this process makes the connection** — + dlt, duckdb, meltano, and the `embedded` modes of debezium and airbyte. For kafka_connect and + the `bring-your-own` / `managed` modes of debezium and airbyte, forge-cli does not connect to + the database at all: it hands the whole connector config, resolved credential included, to the + orchestrator named by `properties..deployment.server_url`. That URL is contract-chosen + and **this policy does not check it**, so a loopback `connection.host` still clears a send whose + credential then leaves the machine in an HTTP body. The behaviour is unchanged from before this + release — nothing regressed — but it is NOT closed, and `deployment.server_url` has no host + allowlist of the kind `FLUID_COMMAND_CENTER_HOST_ALLOWLIST`, `FLUID_FEDERATION_HOST_ALLOWLIST` + and `FLUID_WEBHOOK_HOST_ALLOWLIST` already provide elsewhere. Treat an `env://` secretRef on + those three paths as unprotected whatever the declared host. +- A connection whose only destination field is an object-store URI (`s3://`, `gs://`, `az://`, + `file://`) is not policed: the authority there is a bucket, not a host, so a host-based policy + has nothing to decide. +- `{{ env.NAME }}` interpolation in a connection field reads the environment without going through + `secretRef` at all, so it is not covered by this policy. -### Known limitations - On DuckDB 1.5.x, `lock_configuration` does not hold for logging reached as a table function: `SET enable_logging = true` is refused, but `SELECT * FROM enable_logging(...)` @@ -104,6 +147,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 output port's `query_sql` under `policy.allow_free_form_sql`. DuckDB 2.0 refuses it cleanly. +### Upgrade notes + +- **A contract using `env://` against a remote host now fails until the operator allows it.** Each + one is restored by setting a single operator-environment variable, + `FLUID_ACQUISITION_SECRET_REF_ALLOWLIST`, to the variable's own name — for example + `PG_PASSWORD@db.example.com` for that host only, or `PG_PASSWORD` for any host. Set it in the + shell, in CI, or in a `FLUID_SECRETS_FILE` outside the project — **not** in a project `.env`, + where it is ignored. + A host written with a port never matches, so leave the port off. Local development against + `localhost`, `127.0.0.0/8` or `::1` needs no configuration. The better long-term answer for a + production credential is a secret store: `vault://`, `aws://`, `gcp://`, `azure://` or `file://`. +- `fluid init --discover` against a remote host emits a contract carrying `env://`, which `fluid + apply` will now refuse until that variable is allowlisted. The discover hint does not yet print + the allowlist line; the refusal from `apply` does. + ## [0.22.0] — 2026-10-22 ### Upgrade notes diff --git a/fluid_build/build_runners/_acquisition_common.py b/fluid_build/build_runners/_acquisition_common.py index 931a4bca..e0a5d3f1 100644 --- a/fluid_build/build_runners/_acquisition_common.py +++ b/fluid_build/build_runners/_acquisition_common.py @@ -40,6 +40,11 @@ from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple +from fluid_build.build_runners._secret_policy import ( + refuse_unless_allowed, + register_resolved, +) + LOG = logging.getLogger("fluid.build_runners.acquisition") # Strips ANSI CSI sequences and ASCII control chars from error strings @@ -525,7 +530,27 @@ def resolve_connection_secrets( secret_ref = out.pop("secretRef", None) if not secret_ref: return out + # The policy runs BEFORE the reference is resolved, so a refused send never + # reads the secret. This is the one place that sees both halves the contract + # chose -- the variable and the destination -- which is why the check lives + # here and all six acquisition runners inherit it. + if out.get(target_field): + # An inline literal already wins (below), so the reference is never + # read and nothing is sent anywhere: there is nothing to police. The + # upgrade note promises exactly this, and checking here would refuse a + # contract that has a working credential and a stale secretRef beside + # it. + return out + scheme = str(secret_ref).partition("://")[0].strip().lower() + if scheme == "env": + variable = str(secret_ref).partition("://")[2].strip() + # The whole connection, so EVERY destination it carries has to clear. + refuse_unless_allowed(variable, connection=out, scheme="env") value = resolve_secret_ref(secret_ref) + # #884: the literal can be echoed by a client library's own error, which the + # runners print. Registering it means the redactor masks it by value rather + # than hoping a pattern recognises its shape. + register_resolved(value) if not out.get(target_field): out[target_field] = value return out diff --git a/fluid_build/build_runners/_secret_policy.py b/fluid_build/build_runners/_secret_policy.py new file mode 100644 index 00000000..5b66402b --- /dev/null +++ b/fluid_build/build_runners/_secret_policy.py @@ -0,0 +1,291 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Which environment variable an acquisition source may be sent, and where. + +A contract's ``connection.secretRef: env://NAME`` named the variable to read, +and the same contract named the host the value is sent to as a password. Both +halves were the contract's to choose and nothing checked either, so a contract +run where the environment holds secrets -- CI, an operator's shell -- could send +any variable to a host of its choosing. All six acquisition runners went through +one helper, so all six were affected. + +The policy, chosen by the founder on 2026-10-26 after the options were laid out: + +* A variable is sent to a REMOTE host only if the operator listed it in + :data:`ALLOWLIST_ENV`, as ``NAME`` (any host) or, recommended, ``NAME@host`` + (that hostname only). +* A LOOPBACK host needs no entry. ``env://`` against ``localhost`` is where it + is genuinely convenient -- the helper's own docstring calls it "the hot-path + local-dev case" -- and sending a variable to your own machine discloses it to + nobody. Everything that leaves the machine is checked. +* Anything else is refused before the value is read, so the secret is never + resolved, never placed in a connection dict and never sent. + +This mirrors the federation credential policy in ``forge/federation.py``, which +solved the same problem for ``federation/upstreams.yaml`` and is documented in +``docs/FEDERATION.md``. The one deliberate difference: federation can default +``GITHUB_TOKEN`` to GitHub hosts because a token's destination is self-evident +there. An acquisition source is an arbitrary database at an arbitrary host, so +there is no variable whose destination is obvious and no default allowance. + +:data:`ALLOWLIST_ENV` is an operator variable: it is registered in +``credentials/protected_env.py``, so a value set by one of the project's own +dotenv files is ignored. That matters because ``fluid apply`` hydrates the +environment from ``.env``, ``.env.`` and ``FLUID_SECRETS_FILE``, and those +files are committed beside the contract -- without it, a contract repository +could grant itself the allowlist meant to constrain it. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any, List, Mapping, Optional, Tuple +from urllib.parse import urlsplit + +from fluid_build._errors import FluidUserError, doc_url + +#: Operator-only. Comma-separated ``NAME`` or ``NAME@host`` entries. +ALLOWLIST_ENV = "FLUID_ACQUISITION_SECRET_REF_ALLOWLIST" + +#: Every field that can carry a destination. ALL of them are checked, not just +#: the first that parses: a contract that set ``host: localhost`` as a decoy +#: beside a remote ``connection_string`` would otherwise be exempted on the +#: strength of the decoy. ``connection_string`` is debezium's mongodb shape; +#: ``extra_config`` can carry ``database.hostname`` and friends. +_HOST_FIELDS = ("host", "uri", "instance_url", "connection_string", "server", "endpoint") + +#: URI schemes that name an object store rather than a database server. Their +#: authority is a BUCKET, not a host, so reporting it as a destination both +#: refuses working contracts and suggests an allowlist entry that is nonsense. +_OBJECT_STORE_SCHEMES = frozenset({"s3", "s3a", "s3n", "gs", "gcs", "az", "azure", "abfss", "file"}) + + +@dataclass +class AcquisitionSecretPolicyError(FluidUserError): + """A ``secretRef`` would be sent somewhere the operator has not allowed. + + Raised BEFORE the reference is resolved, so the secret is never read. The + message names the variable and the host and never contains the value. + """ + + code: str = "AcquisitionSecretPolicyError" + + +def connection_hosts(connection: Mapping[str, Any]) -> List[Optional[str]]: + """Every destination this connection block could send a credential to. + + A list, because the policy has to clear ALL of them: trusting the first + field that happened to parse let a contract put ``host: localhost`` beside + a remote ``connection_string`` and be exempted on the decoy. + + An entry is ``None`` when a present field names something we cannot read as + a host, which :func:`refuse_unless_allowed` treats as remote -- a + destination we cannot read is not one we can clear. An object-store URI + (``s3://bucket/...``) is skipped entirely rather than reported as a host: + its authority is a bucket, so refusing on it would break working contracts + and suggest a meaningless allowlist entry. + """ + out: List[Optional[str]] = [] + for field in _HOST_FIELDS: + raw = connection.get(field) + if raw is None or not isinstance(raw, (str, bytes)) or str(raw).strip() == "": + continue + text = str(raw).strip() + if field == "host": + out.append(text) + continue + scheme = text.partition("://")[0].strip().lower() if "://" in text else "" + if scheme in _OBJECT_STORE_SCHEMES: + continue + try: + parsed = urlsplit(text if "//" in text else f"//{text}") + except ValueError: + out.append(None) + continue + out.append(parsed.hostname or None) + return out + + +def connection_host(connection: Mapping[str, Any]) -> Optional[str]: + """The first destination :func:`connection_hosts` found, for diagnostics. + + The POLICY uses ``connection_hosts`` and clears every entry; this is the + one a message names, and the first remote one is the informative choice. + """ + hosts = connection_hosts(connection) + for host in hosts: + if not is_loopback(host): + return host + return hosts[0] if hosts else None + + +def is_loopback(host: Optional[str]) -> bool: + """Whether ``host`` names this machine, so sending it a secret leaves nowhere. + + Delegates to ``_acquisition_common._is_loopback_host`` deliberately rather + than deciding again. The two MUST agree: the exemption below is sound only + because ``apply_loopback_host_override`` rewrites exactly the hosts that + helper calls loopback. A second definition that drifted wider here would + exempt a send the override then leaves pointing at a real host; one that + drifted narrower would refuse a send that never leaves the machine. + + So ``localhost``, the whole of ``127.0.0.0/8`` and ``::1`` are loopback, + and a name like ``db.localhost`` is NOT -- it needs an allowlist entry. + That is stricter than RFC 6761 would allow and it is the right way round: + the override would not rewrite it either, so the contract really would be + reaching whatever that name resolves to. + + No DNS, in either implementation: a name that resolves to loopback now can + resolve elsewhere mid-run, and a policy that depends on a lookup is one an + attacker can move. + """ + from fluid_build.build_runners._acquisition_common import _is_loopback_host + + if host is None: + return False + # A trailing dot is the fully-qualified spelling of the same name, and the + # shared helper does not strip it. + return _is_loopback_host(host.strip().rstrip(".") or host) + + +def _entries(raw: str) -> List[Tuple[str, Optional[str]]]: + """Parse ``NAME`` / ``NAME@host`` entries; malformed ones are dropped.""" + out: List[Tuple[str, Optional[str]]] = [] + for part in raw.split(","): + entry = part.strip() + if not entry: + continue + name, sep, host = entry.partition("@") + name = name.strip() + if not name: + continue + if sep: + target = host.strip().lower().rstrip(".") + if not target: + continue # `NAME@` is malformed: it grants nothing, not everything + out.append((name, target)) + else: + out.append((name, None)) + return out + + +def allowlist(environ: Optional[Mapping[str, str]] = None) -> List[Tuple[str, Optional[str]]]: + """The operator's allowlist as ``(variable, host-or-None)`` pairs. + + Read from the environment, where ``credentials/protected_env.py`` has + already dropped any value a project dotenv file tried to set. + """ + env = os.environ if environ is None else environ + return _entries(env.get(ALLOWLIST_ENV, "") or "") + + +def _allows(variable: str, host: Optional[str], entries: List[Tuple[str, Optional[str]]]) -> bool: + target = (host or "").strip().lower().rstrip(".") + for name, allowed_host in entries: + if name != variable: + continue + if allowed_host is None: + return True # NAME with no host: any destination + if target and allowed_host == target: + return True + return False + + +def refuse_unless_allowed( + variable: str, + *, + connection: Optional[Mapping[str, Any]] = None, + host: Optional[str] = None, + scheme: str = "env", + environ: Optional[Mapping[str, str]] = None, +) -> None: + """Raise unless ``variable`` may be sent to every destination in ``connection``. + + Called before the reference is resolved, so a refusal never reads the + secret. Only ``env://`` is policed: the other schemes name a secret in a + store the operator provisioned and granted this process access to, which is + already a decision the operator made, whereas ``env://`` turns whatever + happens to be in the environment into a credential for a host the contract + picked. + + ``connection`` is the preferred argument: EVERY destination it carries has + to clear, so a decoy loopback field cannot mask a remote sibling. ``host`` + is for the call sites that know their one destination directly (an + orchestrator URL), and is checked on its own. + """ + if scheme != "env": + return + targets: List[Optional[str]] = [] + if connection is not None: + targets.extend(connection_hosts(connection)) + if not targets and any(field in connection for field in _HOST_FIELDS): + # Every destination field present named an object store, whose + # authority is a bucket rather than a host. This host-based policy + # has nothing to decide on, and refusing would break a working + # contract while suggesting a meaningless allowlist entry. A + # KNOWN GAP, recorded in the module docstring: an object-store + # credential is not covered here. + return + if host is not None or not targets: + targets.append(host) + entries = allowlist(environ) + for target in targets: + if is_loopback(target): + continue + if _allows(variable, target, entries): + continue + where = f"host {target!r}" if target else "a host that could not be determined" + suggestion = ( + f'{ALLOWLIST_ENV}="{variable}@{target}"' if target else f'{ALLOWLIST_ENV}="{variable}"' + ) + raise AcquisitionSecretPolicyError( + what=( + f"the contract would send environment variable {variable!r} to {where} " + "as a credential" + ), + why=( + "A contract names both the variable to read and the host it is sent to, so " + "without an operator allowlist any variable in this environment could be " + "sent to a host the contract chose. A loopback host needs no entry; every " + "destination that leaves this machine does." + ), + fix=( + f"If this is intended, set it in the OPERATOR environment (a shell or CI " + f"variable, not a project .env, which is ignored for this setting): " + f'{suggestion} for that host only, or {ALLOWLIST_ENV}="{variable}" for any ' + "host. A host written with a port never matches, so leave the port off. " + "Otherwise put the credential in a secret store and reference it with " + "vault://, aws://, gcp://, azure:// or file://." + ), + doc=doc_url(), + ) + + +def register_resolved(value: Any) -> None: + """Register a resolved secret with the exact-value redactor (#884). + + A value that reached a connection dict can be echoed by a client library's + own error, and the runners print those. Registering the literal means the + redactor masks it wherever it later appears, rather than relying on the + pattern rules to recognise its shape. Never fatal: failing to register must + not fail a run that is otherwise fine. + """ + try: + from fluid_build.observability.secret_redactor import register_secret + + register_secret(value) + except Exception: # pragma: no cover - redaction is best-effort + return diff --git a/fluid_build/build_runners/airbyte/runner.py b/fluid_build/build_runners/airbyte/runner.py index 789914fc..21803205 100644 --- a/fluid_build/build_runners/airbyte/runner.py +++ b/fluid_build/build_runners/airbyte/runner.py @@ -52,6 +52,12 @@ utc_now_iso, write_run_record_and_finalize, ) +from .._secret_policy import ( + AcquisitionSecretPolicyError, + connection_host, + refuse_unless_allowed, + register_resolved, +) LOG = logging.getLogger("fluid.acquire.airbyte") @@ -306,7 +312,20 @@ def _execute_rest_mode( api_token = auth.get("token") if not api_token and auth.get("secretRef"): try: + # The contract names deployment.server_url, so this reference takes + # the same policy as a source connection's -- the destination here + # is that URL's host, not a connection block's. + _ref = str(auth["secretRef"]) + if _ref.partition("://")[0].strip().lower() == "env": + refuse_unless_allowed( + _ref.partition("://")[2].strip(), + host=connection_host({"uri": server_url}), + scheme="env", + ) api_token = resolve_secret_ref(auth["secretRef"]) + register_resolved(api_token) + except AcquisitionSecretPolicyError: + raise except (ValueError, NotImplementedError) as exc: return _failed( ctx, diff --git a/fluid_build/build_runners/duckdb/runner.py b/fluid_build/build_runners/duckdb/runner.py index cf20293a..203add1b 100644 --- a/fluid_build/build_runners/duckdb/runner.py +++ b/fluid_build/build_runners/duckdb/runner.py @@ -58,6 +58,7 @@ ) from fluid_build.api.schema import SchemaFingerprint from fluid_build.api.source import AcquisitionMode +from fluid_build.build_runners._secret_policy import AcquisitionSecretPolicyError from fluid_build.credentials.host_db_env import host_credential_env_suppressed from fluid_build.providers._duckdb_compat import first_row from fluid_build.providers._duckdb_sandbox import ( @@ -864,6 +865,11 @@ def _attach_external_databases(self, con: Any, ctx: RunContext) -> None: # Resolve secretRef → password before the DSN is built; an # inline literal still wins. conn = resolve_connection_secrets(dict(ctx.source.connection.raw)) + except AcquisitionSecretPolicyError: + # A typed user error: let the CLI render its what/why/fix panel + # rather than flattening it to a one-line _SourceSetupError, + # which drops the allowlist line the operator needs. + raise except Exception as exc: # An unset variable or an unknown scheme (ValueError), or the # secret backend's own error (ConfigurationError from vault, diff --git a/fluid_build/credentials/protected_env.py b/fluid_build/credentials/protected_env.py index f2194047..b81d0916 100644 --- a/fluid_build/credentials/protected_env.py +++ b/fluid_build/credentials/protected_env.py @@ -268,6 +268,14 @@ "FLUID_COMMAND_CENTER_HOST_ALLOWLIST", "FLUID_FEDERATION_HOST_ALLOWLIST", "FLUID_FEDERATION_SECRET_REF_ALLOWLIST", + "FLUID_ACQUISITION_SECRET_REF_ALLOWLIST", + # Both rewrite a contract-declared loopback host to a reachable one, + # AFTER the secretRef policy has cleared the send as local-only + # (build_runners/_secret_policy). A project file that could set either + # would redirect a credential the policy allowed only because it was + # staying on this machine. FLUID_RUNNER_HOST_OVERRIDE was already + # listed; the testcontainers name it also honours was not. + "TESTCONTAINERS_HOST_OVERRIDE", "FLUID_WEBHOOK_HOST_ALLOWLIST", "DMM_ALLOW_INSECURE_HTTP", "DMM_AUTO_APPROVE_ACCESS", diff --git a/tests/build_runners/test_acquisition_common.py b/tests/build_runners/test_acquisition_common.py index 42d617f3..7039591b 100644 --- a/tests/build_runners/test_acquisition_common.py +++ b/tests/build_runners/test_acquisition_common.py @@ -111,6 +111,21 @@ def test_scheme_is_case_insensitive(self, monkeypatch: pytest.MonkeyPatch): class TestResolveConnectionSecrets: """Convenience wrapper that places the resolved secret into a target field.""" + @pytest.fixture(autouse=True) + def _allow_the_remote_host(self, monkeypatch: pytest.MonkeyPatch) -> None: + """These cases exercise RESOLUTION, not the send policy. + + ``env://`` to a remote host now needs the operator's allowlist + (``build_runners/_secret_policy``), and ``db.example`` is remote. Listing + the variables keeps these tests about what they were about, and is the + same one-line change an operator makes on upgrade. The policy itself is + covered in ``test_acquisition_secret_policy.py``. + """ + monkeypatch.setenv( + "FLUID_ACQUISITION_SECRET_REF_ALLOWLIST", + "PG_PASSWORD,API_TOKEN,TOKEN,SECRET,MY_SECRET,GH_TOKEN", + ) + def test_no_secret_ref_returns_shallow_copy_unchanged(self): original = {"host": "db.example", "port": 5432, "user": "alice"} out = resolve_connection_secrets(original) diff --git a/tests/build_runners/test_acquisition_secret_policy.py b/tests/build_runners/test_acquisition_secret_policy.py new file mode 100644 index 00000000..067fa330 --- /dev/null +++ b/tests/build_runners/test_acquisition_secret_policy.py @@ -0,0 +1,381 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""A contract cannot send any environment variable to a host of its choosing. + +``connection.secretRef: env://NAME`` let the contract name BOTH halves -- the +variable to read and the host the value is sent to as a password -- and nothing +checked either. A contract run where the environment holds secrets (CI, an +operator's shell) could therefore exfiltrate any variable to a host it picked. +All six acquisition runners go through one helper, so all six were affected. + +Policy (founder, 2026-10-26): a remote host needs the variable listed in +``FLUID_ACQUISITION_SECRET_REF_ALLOWLIST``; a loopback host needs nothing; the +refusal happens BEFORE the reference resolves, so a refused send never reads +the secret. +""" + +from __future__ import annotations + +import pathlib +from typing import Any, Dict + +import pytest + +from fluid_build.build_runners._acquisition_common import resolve_connection_secrets +from fluid_build.build_runners._secret_policy import ( + ALLOWLIST_ENV, + AcquisitionSecretPolicyError, + connection_host, + is_loopback, + refuse_unless_allowed, +) + +VARIABLE = "FLUID_TEST_PG_PASSWORD" +SECRET = "s3cret-value-not-for-remote-hosts" # pragma: allowlist secret + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(VARIABLE, SECRET) + monkeypatch.delenv(ALLOWLIST_ENV, raising=False) + + +def _resolve(connection: Dict[str, Any]) -> Dict[str, Any]: + return resolve_connection_secrets(dict(connection)) + + +# ── loopback needs no configuration ────────────────────────────────────── + + +@pytest.mark.parametrize( + "connection", + [ + {"host": "localhost"}, + {"host": "127.0.0.1"}, + {"host": "127.5.5.5"}, # the whole /8, not just .1 + {"host": "::1"}, + {"host": "[::1]"}, + {"uri": "postgresql://alice@localhost:5432/sales"}, + ], +) +def test_a_loopback_destination_is_allowed_unconfigured(connection: Dict[str, Any]) -> None: + """Sending a variable to this machine discloses it to nobody.""" + out = _resolve({**connection, "secretRef": f"env://{VARIABLE}"}) + assert out["password"] == SECRET + assert "secretRef" not in out + + +# ── anything leaving the machine is checked ────────────────────────────── + + +@pytest.mark.parametrize( + "connection, why", + [ + ({"host": "db.example.com"}, "a plain remote host"), + ({"uri": "postgresql://alice@db.example.com:5432/sales"}, "a remote host in a uri"), + ({"instance_url": "https://corp.example.com"}, "a remote host in instance_url"), + ({}, "no host at all: a destination we cannot read is not one we can clear"), + ({"host": "db.localhost"}, "a .localhost NAME still resolves wherever DNS says"), + ], +) +def test_a_remote_destination_is_refused_unconfigured(connection: Dict[str, Any], why: str) -> None: + with pytest.raises(AcquisitionSecretPolicyError) as caught: + _resolve({**connection, "secretRef": f"env://{VARIABLE}"}) + assert VARIABLE in caught.value.what, why + assert SECRET not in str(caught.value.what) + str(caught.value.why) + str(caught.value.fix) + + +def test_the_refusal_never_contains_the_secret(monkeypatch: pytest.MonkeyPatch) -> None: + """The whole rendered error, not just one field.""" + with pytest.raises(AcquisitionSecretPolicyError) as caught: + _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + rendered = " ".join( + str(getattr(caught.value, field, "")) for field in ("what", "why", "fix", "doc", "code") + ) + assert SECRET not in rendered + assert ALLOWLIST_ENV in rendered, "the fix has to say how to allow it" + + +def test_the_secret_is_not_read_when_the_send_is_refused( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Refuse first, resolve second: a refused reference must never be read. + + Checked by removing the variable entirely — if resolution were attempted it + would raise ValueError("environment variable not set") instead, and the + policy error is what proves the order. + """ + monkeypatch.delenv(VARIABLE, raising=False) + with pytest.raises(AcquisitionSecretPolicyError): + _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + + +# ── the operator's allowlist ───────────────────────────────────────────── + + +def test_a_listed_variable_and_host_is_allowed(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ALLOWLIST_ENV, f"{VARIABLE}@db.example.com") + out = _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + assert out["password"] == SECRET + + +def test_a_listing_for_another_host_does_not_transfer(monkeypatch: pytest.MonkeyPatch) -> None: + """``NAME@host`` is the recommended form precisely because it is per-host.""" + monkeypatch.setenv(ALLOWLIST_ENV, f"{VARIABLE}@db.example.com") + with pytest.raises(AcquisitionSecretPolicyError): + _resolve({"host": "other.example.com", "secretRef": f"env://{VARIABLE}"}) + + +def test_a_bare_name_allows_any_host(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ALLOWLIST_ENV, VARIABLE) + out = _resolve({"host": "anywhere.example.com", "secretRef": f"env://{VARIABLE}"}) + assert out["password"] == SECRET + + +def test_another_variables_listing_does_not_cover_this_one( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(ALLOWLIST_ENV, "SOMETHING_ELSE@db.example.com") + with pytest.raises(AcquisitionSecretPolicyError): + _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + + +@pytest.mark.parametrize( + "raw", + [ + f" {VARIABLE}@db.example.com ", + f"OTHER,{VARIABLE}@db.example.com", + f"{VARIABLE}@DB.EXAMPLE.COM", # host comparison is case-insensitive + f",,{VARIABLE}@db.example.com,,", + ], +) +def test_the_allowlist_is_parsed_forgivingly(raw: str, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(ALLOWLIST_ENV, raw) + assert _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"})["password"] + + +def test_a_malformed_entry_grants_nothing(monkeypatch: pytest.MonkeyPatch) -> None: + """Fail closed: an entry we cannot read is not an entry that allows.""" + monkeypatch.setenv(ALLOWLIST_ENV, "@db.example.com, ,@") + with pytest.raises(AcquisitionSecretPolicyError): + _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + + +# ── scope: what the policy does NOT touch ──────────────────────────────── + + +def test_an_inline_password_is_untouched() -> None: + """No secretRef, no policy: the contract carried its own credential.""" + out = _resolve({"host": "db.example.com", "password": "inline"}) # pragma: allowlist secret + assert out["password"] == "inline" # pragma: allowlist secret + + +def test_a_store_backed_scheme_is_not_policed(monkeypatch: pytest.MonkeyPatch) -> None: + """vault:// etc. name a secret the OPERATOR provisioned and granted access to. + + That is already an operator decision, unlike ``env://``, which turns + whatever happens to be in the environment into a credential for a host the + contract picked. Checked at the policy layer so the test needs no backend. + """ + refuse_unless_allowed("ANYTHING", host="db.example.com", scheme="vault") + refuse_unless_allowed("ANYTHING", host="db.example.com", scheme="aws") + + +# ── the premise the loopback exemption rests on ────────────────────────── + + +def test_the_policy_and_the_loopback_override_agree() -> None: + """The exemption is only sound while these two cover the SAME hosts. + + ``apply_loopback_host_override`` rewrites a contract-declared loopback host + to a reachable one AFTER this policy has cleared the send. If the policy + called a host loopback that the override does not rewrite, nothing moves and + the exemption is honest; if the policy were WIDER than the override, a send + cleared as local-only would end up pointing at a real host. One definition, + asserted here so a future edit to either cannot drift. + """ + from fluid_build.build_runners._acquisition_common import _is_loopback_host + + for host in ( + "localhost", + "127.0.0.1", + "127.5.5.5", + "::1", + "[::1]", + "db.localhost", + "db.example.com", + "", + None, + ): + assert is_loopback(host) is _is_loopback_host(host), host + + +def test_both_override_variables_are_operator_only() -> None: + """Either one redirects a loopback-exempted credential, so neither may come + from a project dotenv file committed beside the contract.""" + from fluid_build.credentials.protected_env import is_protected_env_key + + assert is_protected_env_key("FLUID_RUNNER_HOST_OVERRIDE") + assert is_protected_env_key("TESTCONTAINERS_HOST_OVERRIDE") + + +def test_the_allowlist_itself_is_operator_only() -> None: + """`fluid apply` hydrates .env files committed beside the contract, so a + project file must not be able to grant the allowlist that constrains it.""" + from fluid_build.credentials.protected_env import is_protected_env_key + + assert is_protected_env_key(ALLOWLIST_ENV) + + +# ── host extraction ────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "connection, expected", + [ + ({"host": "db.example.com"}, "db.example.com"), + ({"host": "db.example.com", "uri": "postgresql://ignored/"}, "db.example.com"), + # a userinfo-bearing URI is the shape under test + ( + {"uri": "postgresql://alice:pw@db.example.com:5432/sales"}, # pragma: allowlist secret + "db.example.com", + ), + ({"uri": "db.example.com:5432"}, "db.example.com"), + ({"uri": "postgresql://alice@[2001:db8::1]:5432/s"}, "2001:db8::1"), + ({"instance_url": "https://corp.example.com/api"}, "corp.example.com"), + ({}, None), + ({"host": " "}, None), + ], +) +def test_the_destination_is_read_from_the_connection( + connection: Dict[str, Any], expected: Any +) -> None: + assert connection_host(connection) == expected + + +# ── the resolved value reaches the redactor (#884) ─────────────────────── + + +def test_a_resolved_secret_is_registered_with_the_redactor( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A client library can echo the literal in its own error, which runners print.""" + seen = [] + import fluid_build.observability.secret_redactor as redactor + + monkeypatch.setattr(redactor, "register_secret", lambda v: seen.append(v) or True) + monkeypatch.setenv(ALLOWLIST_ENV, VARIABLE) + _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + assert SECRET in seen, "the resolved value was not registered for redaction" + + +# ── what the adversarial review of this policy found ──────────────────── + + +def test_a_decoy_loopback_field_cannot_mask_a_remote_sibling() -> None: + """The HIGH finding: trusting the first field that parses was a bypass. + + A contract set ``host: localhost`` beside a remote ``connection_string`` + (debezium's mongodb shape) and was exempted on the decoy, while the + credential went to the remote one. EVERY destination has to clear. + """ + with pytest.raises(AcquisitionSecretPolicyError) as caught: + _resolve( + { + "host": "localhost", + "connection_string": "mongodb://evil.example.com/db", + "secretRef": f"env://{VARIABLE}", + } + ) + assert "evil.example.com" in caught.value.what, caught.value.what + + +@pytest.mark.parametrize( + "field", ["connection_string", "server", "endpoint", "uri", "instance_url"] +) +def test_every_destination_bearing_field_is_read(field: str) -> None: + """A field the policy does not read is a field a contract can hide behind.""" + with pytest.raises(AcquisitionSecretPolicyError): + _resolve({field: "remote.example.com", "secretRef": f"env://{VARIABLE}"}) + + +def test_a_trailing_at_grants_nothing(monkeypatch: pytest.MonkeyPatch) -> None: + """``NAME@`` parsed fail-OPEN to any host, which is the opposite of intended.""" + monkeypatch.setenv(ALLOWLIST_ENV, f"{VARIABLE}@") + with pytest.raises(AcquisitionSecretPolicyError): + _resolve({"host": "evil.example.com", "secretRef": f"env://{VARIABLE}"}) + + +def test_an_inline_credential_is_not_policed() -> None: + """The upgrade note promises this: the reference is never read, so nothing is sent.""" + out = _resolve( + { + "host": "db.example.com", + "password": "inline", # pragma: allowlist secret + "secretRef": f"env://{VARIABLE}", + } + ) + assert out["password"] == "inline" # pragma: allowlist secret + assert "secretRef" not in out + + +def test_an_object_store_uri_is_not_treated_as_a_host() -> None: + """Its authority is a bucket, so a host policy has nothing to decide. + + Refusing it broke working contracts AND suggested an allowlist entry naming + a bucket, which is meaningless. Recorded as a known limitation rather than + silently cleared. + """ + for uri in ("s3://bucket/prefix", "gs://bucket/p", "az://c/p", "file:///tmp/x"): + out = _resolve({"uri": uri, "secretRef": f"env://{VARIABLE}"}) + assert out["password"] == SECRET, uri + + +def test_the_fqdn_form_of_localhost_is_loopback() -> None: + assert is_loopback("localhost.") + assert _resolve({"host": "localhost.", "secretRef": f"env://{VARIABLE}"})["password"] == SECRET + + +def test_the_refusal_carries_its_own_error_code() -> None: + """Without @dataclass the inherited `code` read "FluidUserError".""" + with pytest.raises(AcquisitionSecretPolicyError) as caught: + _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + assert caught.value.code == "AcquisitionSecretPolicyError" + + +def test_the_fix_warns_that_a_port_never_matches() -> None: + """federation already warns about this; the message should pre-empt it.""" + with pytest.raises(AcquisitionSecretPolicyError) as caught: + _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + assert "port" in caught.value.fix + + +def test_a_host_written_with_a_port_does_not_match(monkeypatch: pytest.MonkeyPatch) -> None: + """Fails closed, as federation does, rather than appearing to work.""" + monkeypatch.setenv(ALLOWLIST_ENV, f"{VARIABLE}@db.example.com:5432") + with pytest.raises(AcquisitionSecretPolicyError): + _resolve({"host": "db.example.com", "secretRef": f"env://{VARIABLE}"}) + + +def test_the_airbyte_deployment_token_path_is_policed() -> None: + """It was claimed as policed in an earlier commit message and was not. + + The edit had failed on a bad anchor and the omission went unnoticed, so this + pins the call site rather than the claim. + """ + from fluid_build.build_runners.airbyte import runner as airbyte_runner + + source = pathlib.Path(airbyte_runner.__file__).read_text(encoding="utf-8") + assert "refuse_unless_allowed(" in source, "the deployment.auth.secretRef path is unpoliced" diff --git a/tests/build_runners/test_duckdb_runner_mysql_attach.py b/tests/build_runners/test_duckdb_runner_mysql_attach.py index fe9b31c4..2cdc5f96 100644 --- a/tests/build_runners/test_duckdb_runner_mysql_attach.py +++ b/tests/build_runners/test_duckdb_runner_mysql_attach.py @@ -40,6 +40,21 @@ } +@pytest.fixture(autouse=True) +def _allow_the_fixture_host(monkeypatch: pytest.MonkeyPatch) -> None: + """``env://`` to a remote host needs the operator's allowlist now. + + These suites exercise the ATTACH and its error handling, not the secretRef + send policy, and the fixture host (``db.test``) is remote. Listing the two + variables they use keeps them about what they were about; the policy itself + is covered in ``test_acquisition_secret_policy.py``. + """ + monkeypatch.setenv( + "FLUID_ACQUISITION_SECRET_REF_ALLOWLIST", + "RUNNER_FAKE_PG_SECRET,RUNNER_FAKE_MYSQL_SECRET,RUNNER_FAKE_UNSET_SECRET", + ) + + def _attach_on(con: Any, raw: Dict[str, Any], kind: str = "mysql") -> None: """Run the runner's attach of a ``kind`` source with ``raw`` on ``con``.""" ctx = types.SimpleNamespace( diff --git a/tests/security/test_credentials_stay_out_of_files_and_diagnostics.py b/tests/security/test_credentials_stay_out_of_files_and_diagnostics.py index 4ab34afd..166820de 100644 --- a/tests/security/test_credentials_stay_out_of_files_and_diagnostics.py +++ b/tests/security/test_credentials_stay_out_of_files_and_diagnostics.py @@ -537,6 +537,12 @@ def test_embedded_server_file_holds_no_password_and_server_gets_it(tmp_path, mon from fluid_build.build_runners.debezium import runner as dbz monkeypatch.setenv("PG_PASSWORD_T211", _SECRET) + # `env://` to a remote host needs the operator's allowlist now + # (build_runners/_secret_policy); this fixture's host is the bare name `db`. + # This suite is about the secret not reaching a FILE or a diagnostic once it + # is legitimately in use, which is a different control from whether the send + # is permitted at all. + monkeypatch.setenv("FLUID_ACQUISITION_SECRET_REF_ALLOWLIST", "PG_PASSWORD_T211") monkeypatch.setenv("LAKEKEEPER_CREDENTIAL_T211", "inherited") captured = tmp_path / "server-env.txt" server = tmp_path / "debezium-server"