From eaf6a4868dcc547307a5e9b58da0d28350257f56 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 20:06:43 +0100 Subject: [PATCH 1/2] security: a contract could send a product's credential to any host it named MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On the kafka-connect path, and on the DEFAULT `bring-your-own` / `managed` modes of debezium and airbyte, forge-cli never connects to the source database. It resolves the contract's `connection.secretRef`, puts the plaintext into the connector config (`database.password`, `connection.password`) and POSTs the whole config to `properties..deployment.server_url` — a URL the CONTRACT chooses. Nothing constrained that host, so the credential could go anywhere, needing no allowlist entry and no operator mistake; a loopback `connection.host` even made the build look local, so the `env://` policy from #751 cleared it as staying on this machine. A public orchestrator host must now be named in `FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST`. Loopback and private hosts need no entry — the ordinary cluster deployment, and a credential that stays inside the organisation is not the leak this closes. Link-local and unspecified are refused outright, allowlist or not: that is where instance metadata lives. **The three existing HOST_ALLOWLIST variables are the wrong model.** The card said to mirror `FLUID_COMMAND_CENTER_HOST_ALLOWLIST`, `FLUID_FEDERATION_HOST_ALLOWLIST` and `FLUID_WEBHOOK_HOST_ALLOWLIST`. All three are SSRF allowlists: unset, they PERMIT any public host and refuse private ones, and the Command Center gate still lets a credential reach any public `https` endpoint unlisted. Mirroring them would have reproduced the hole. The deny-unless-named shape comes from `_secret_policy.py`, which answers the same question for `connection.secretRef`. **A mixed DNS answer is public.** `_net.hostname_scope` reports `public`, `non_public`, `mixed` or `unresolved`, and examines every address. Stopping at the first non-public one — which the first version of this did — was a complete bypass: an attacker naming the host answers with one private record beside their public one, the policy calls the host internal and allows it unlisted, and `safe_http` pins the public address anyway. `_hostname_is_private` keeps answering True for `mixed` and `unresolved`, its fail-closed direction and the opposite of this gate's, which is why the states cannot be collapsed. **The address predicate and the non-routable-range list move to tier-0 `_net` and `safe_http` re-exports them.** They existed in both places and disagreed: `_hostname_is_private` judged on bare `ipaddress` flags, so a Tailscale/CGNAT address (`100.64.0.0/10`) was public there and non-public in `safe_http`. This policy therefore refused a CGNAT orchestrator as "public" — wrong answer, wrong reason. One list now, consulted by both; `_hostname_is_private` gets stricter for CGNAT/6to4/NAT64/TEST-NET, which only ever refuses more. **The gate runs where each runner reads `server_url`,** before the connector config is built — and building it is what resolves the `secretRef`. The client constructors keep the same check as the chokepoint for any other caller, which is also where an operator-supplied URL is distinguished. The constructor check ALONE was not enough and the first version of this promised an ordering it did not have: all three runners build the config first, so a refused deployment had already fetched the password from Vault/KMS. **A refusal is no longer swallowed.** The kafka-connect status poll caught it as eventual consistency and retried to the deadline; the outer handler flattened it to `str(exc)`, dropping the `why` and the `fix` that names the variable. Both re-raise it now. `fluid import airbyte --server-url` exposed an over-reach in the live run: that URL is the OPERATOR's, and the policy exists because a CONTRACT can choose the destination. It refused a legitimate action and told the operator to set an allowlist variable to permit what they had just asked for. `operator_supplied=True` marks that path, with a message that blames no contract and does not offer the allowlist as the remedy for a metadata address. Modular by request: one policy module, two client constructors (debezium shares kafka-connect's), three runner call sites, so a fourth engine costs one line. The variable is operator-only in `credentials/protected_env.py`, because `fluid apply` hydrates `.env`, `.env.` and `FLUID_SECRETS_FILE`, committed beside the contract. Validation. 56 tests, DNS faked throughout — the first version resolved `example.com` and skipped on failure, so six of eight DNS-dependent assertions could pass without testing anything. Five mechanisms are mutation-checked: first-hit classification, the unconsulted deny-list, the missing `mixed` branch, the removed runner-level gate and a no-op request guard each fail their own test. Full suite CI-shaped on both floors, run sequentially: 27989 and 27988 passed, 4 failed — the same four that fail on clean base. Live: `fluid import airbyte --server-url https://example.com` passes the gate and fails on the real endpoint; the same command against 169.254.169.254 is refused; a contract-chosen public host is refused with exit 1. ruff, black==24.10.0, license headers, import-linter (6 kept / 0 broken), startup budget and detect-secrets all clean. --- CHANGELOG.md | 62 ++ fluid_build/_net.py | 149 ++++- .../build_runners/_deployment_host_policy.py | 271 +++++++++ fluid_build/build_runners/airbyte/runner.py | 30 + fluid_build/build_runners/debezium/runner.py | 12 +- .../build_runners/kafka_connect/runner.py | 56 +- fluid_build/cli/import_workflow/airbyte.py | 12 +- fluid_build/credentials/protected_env.py | 1 + fluid_build/util/safe_http.py | 64 +-- .../test_deployment_host_policy.py | 530 ++++++++++++++++++ .../test_airbyte_server_url_required.py | 9 + 11 files changed, 1114 insertions(+), 82 deletions(-) create mode 100644 fluid_build/build_runners/_deployment_host_policy.py create mode 100644 tests/build_runners/test_deployment_host_policy.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 044e8b23..91fab6e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,68 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Security + +- **A contract could have a product's database credential delivered to any host it named.** On the + `kafka-connect` path, and on the **default** `bring-your-own` / `managed` modes of `debezium` and + `airbyte`, forge-cli never connects to the source database: it resolves the contract's + `connection.secretRef`, puts the plaintext into the connector config (`database.password`, + `connection.password`) and POSTs the whole config to + `properties..deployment.server_url` — a URL the **contract** chooses. Nothing constrained + that host, so the credential could be sent anywhere, needing no allowlist entry and no operator + mistake; a loopback `connection.host` even made the build look local. A **public** orchestrator + host must now be named in `FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST`. + + Loopback and private hosts need no entry: an orchestrator on the cluster network is the ordinary + deployment, a credential that stays inside the organisation is not the leak this closes, and + refusing it would have broken every working install on upgrade. A Tailscale or other CGNAT + address (`100.64.0.0/10`) counts as private, which it did not at first — the extended + non-routable-range list the tree already had was consulted by `util/safe_http` but not by the + classifier this policy uses, so such an orchestrator was called public and refused with the wrong + reason. Both now read one list, in the tier-0 `_net` leaf. + + Link-local and unspecified addresses are refused outright, allowlist or not, because that is where + cloud instance metadata lives. A host that resolves to **both** public and non-public addresses is + treated as public: one private record beside a public one is how an unnamed host is made to look + internal, and the connection is pinned to the public address regardless. + + The gate runs where each runner reads `deployment.server_url`, which is **before** the connector + config is built — and building it is what resolves the `secretRef` — so a refused deployment never + fetches the password from its secret store. The REST client constructors keep the same check as + the chokepoint for any other caller, and it is re-applied per request, so a host that did not + resolve at construction is judged for real once it does. + + A URL the **operator** supplied — `fluid import airbyte --server-url`, or + `FLUID_IMPORT_AIRBYTE_URL` — is not second-guessed, since the policy exists because a *contract* + can choose where a credential goes. The scheme and metadata-address rules still apply there. + + The variable is operator-only (`credentials/protected_env.py`): `fluid apply` hydrates the + environment from `.env`, `.env.` and `FLUID_SECRETS_FILE`, which are committed beside the + contract, so without that a contract repository could grant itself the allowlist meant to + constrain it. + + Deliberately **not** modelled on `FLUID_WEBHOOK_HOST_ALLOWLIST`, + `FLUID_FEDERATION_HOST_ALLOWLIST` or `FLUID_COMMAND_CENTER_HOST_ALLOWLIST` despite the matching + name. Those are SSRF allowlists: unset, they *permit* any public host and refuse private ones, and + the Command Center gate still lets a credential reach any public `https` endpoint unlisted. + Mirroring them would have reproduced the hole. + +### Upgrade notes + +- **A contract whose orchestrator is on a public host now fails until the operator allows it.** + Set `FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST` in the shell, in CI, or in a + `FLUID_SECRETS_FILE` outside the project — **not** in a project `.env`, where it is ignored — to + the host or its parent domain (`connect.example.com`, or `example.com` to cover subdomains). A + host written with a port never matches, so leave the port off. Deployments on `localhost`, + `127.0.0.0/8`, `::1` or a private network need no configuration, which is the common case. + +### Known limitations + +- Two credential paths this policy does not reach, both pre-existing. An object-store URI can carry + its own credentials in the URI itself, which is not a `secretRef` and so is not policed; and + `{{ env.NAME }}` interpolation inside a contract bypasses `secretRef` entirely. Neither is changed + by this release. + ## [0.24.0] — 2026-10-27 ### Fixed diff --git a/fluid_build/_net.py b/fluid_build/_net.py index 4bc94b04..f853527d 100644 --- a/fluid_build/_net.py +++ b/fluid_build/_net.py @@ -31,6 +31,125 @@ import ipaddress import socket +# CIDR ranges ``ipaddress.is_private`` does NOT flag but which are nonetheless +# non-routable / not safe to fetch from. Borrowed from requests-hardened +# (Saleor, BSD-3). Lives here, at tier-0, because BOTH the broad SSRF gate below +# and ``util.safe_http.is_public_address`` judge addresses and must agree: they +# did not before, and the acquisition deployment-host policy -- which allows +# internal hosts and questions public ones -- classified a Tailscale/CGNAT +# orchestrator as public and refused it with the wrong reason. +EXTRA_BLOCKED_NETWORKS = tuple( + ipaddress.ip_network(cidr) + for cidr in ( + "100.64.0.0/10", # RFC 6598 — Carrier-Grade NAT (Tailscale, many ISPs) + "192.0.0.0/24", # RFC 6890 — IETF Protocol Assignments (CPython #113171 gap) + "192.0.2.0/24", # RFC 5737 — TEST-NET-1 + "192.88.99.0/24", # RFC 7526 — 6to4 relay anycast (deprecated) + "198.18.0.0/15", # RFC 2544 — benchmarking + "198.51.100.0/24", # RFC 5737 — TEST-NET-2 + "203.0.113.0/24", # RFC 5737 — TEST-NET-3 + "240.0.0.0/4", # RFC 1112 — class E reserved + "2001:20::/28", # RFC 7343 — ORCHIDv2 + "2002::/16", # RFC 3056 — 6to4 + "5f00::/16", # RFC 9602 — IPv6 Segment Routing + "64:ff9b::/96", # RFC 6052 — Well-Known NAT64 + "64:ff9b:1::/48", # RFC 8215 — Local-Use NAT64 + ) +) + + +def is_public_address(addr: str) -> bool: + """True iff ``addr`` is a routable public IP (v4 or v6). + + Handles three traps ``ipaddress`` alone does not: + * IPv4-mapped IPv6 (``::ffff:a.b.c.d``) — on Python <3.12 the v6 form does + NOT recurse ``is_private`` / ``is_loopback`` into the embedded v4 + (gh-87105 / bpo-44904, fixed only in 3.12). + * Zone-id-bearing v6 (``fe80::1%eth0``) — strip the scope. + * Extended CIDR deny-list (:data:`EXTRA_BLOCKED_NETWORKS`). + + An address that does not parse is not public. ``util.safe_http`` re-exports + this; it used to own it, and the two judgements must not drift. + """ + if "%" in addr: + addr = addr.split("%", 1)[0] + try: + ip = ipaddress.ip_address(addr) + except ValueError: + return False + if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: + ip = ip.ipv4_mapped + if ( + ip.is_private + or ip.is_loopback + or ip.is_link_local + or ip.is_multicast + or ip.is_reserved + or ip.is_unspecified + ): + return False + for network in EXTRA_BLOCKED_NETWORKS: + if ip.version == network.version and ip in network: + return False + return True + + +#: Every host resolves into exactly one of these. +#: +#: ``"public"`` — resolved, and EVERY address is publicly routable. +#: ``"non_public"`` — resolved, and EVERY address is loopback, private, +#: link-local, unspecified, reserved or on the extended +#: deny-list above. +#: ``"mixed"`` — resolved to BOTH. Not a curiosity: an attacker who names +#: the host answers with one private record beside their +#: public one, and a gate that stops at the first non-public +#: address calls the host internal while the connection is +#: pinned to the public one. +#: ``"unresolved"`` — DNS failed, or an address came back unparseable. +HOST_SCOPES = ("public", "non_public", "mixed", "unresolved") + + +def hostname_scope(hostname: str) -> str: + """Which of :data:`HOST_SCOPES` ``hostname`` falls into. + + EVERY address in the answer is examined, and a mixed answer is reported as + such. :func:`_hostname_is_private` cannot express that: it answers one + question ("may I refuse this?") for which ``mixed`` and ``non_public`` are + the same, while a gate with the opposite polarity -- one that ALLOWS + internal hosts and asks the operator to name public ones -- must treat a + mixed answer as public. Collapsing the two is a complete bypass of such a + gate: one decoy AAAA record and the credential goes to the public A record + ``safe_http`` then pins. Found by review of the deployment-host policy and + reproduced before this existed. + + Kept here, beside :func:`is_public_address` and the range list it consults, + rather than re-derived by a caller: this module's whole point is that the + blocked set is written once. + """ + try: + addresses = socket.getaddrinfo(hostname, None) + except socket.gaierror: + return "unresolved" + saw_public = False + saw_non_public = False + for entry in addresses: + ip_str = entry[4][0] + try: + ipaddress.ip_address(ip_str.split("%", 1)[0]) + except ValueError: + return "unresolved" + if is_public_address(ip_str): + saw_public = True + else: + saw_non_public = True + if saw_public and saw_non_public: + return "mixed" + if saw_non_public: + return "non_public" + # No addresses at all lands here, as it did before: the old loop fell + # through to "not private". + return "public" + def _hostname_is_private(hostname: str) -> bool: """Return True when ``hostname`` resolves to a non-public IP. @@ -59,26 +178,18 @@ def _hostname_is_private(hostname: str) -> bool: depend on it without re-introducing the ``observability → build_runners → cli → observability`` cycle that previously broke ``cli/__init__.py`` import-time. + + Expressed in terms of :func:`hostname_scope` so the blocked range list is + written once. ``unresolved``, ``non_public`` and ``mixed`` all answer True, + which is what the old "any non-public address" loop did. + + One deliberate change: the extended deny-list is now consulted here too, so + a CGNAT / 6to4 / NAT64 / TEST-NET host answers True where it used to answer + False. That only ever refuses more, never allows more, and it matches what + ``safe_http.resolve_and_validate`` already did for the same addresses -- + the two disagreed before. """ - try: - addresses = socket.getaddrinfo(hostname, None) - except socket.gaierror: - return True - for entry in addresses: - ip_str = entry[4][0] - try: - ip = ipaddress.ip_address(ip_str) - except ValueError: - return True - if ( - ip.is_loopback - or ip.is_private - or ip.is_link_local - or ip.is_unspecified - or ip.is_reserved - ): - return True - return False + return hostname_scope(hostname) != "public" def _hostname_is_link_local(hostname: str) -> bool: diff --git a/fluid_build/build_runners/_deployment_host_policy.py b/fluid_build/build_runners/_deployment_host_policy.py new file mode 100644 index 00000000..604157bd --- /dev/null +++ b/fluid_build/build_runners/_deployment_host_policy.py @@ -0,0 +1,271 @@ +# 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. + +"""Where an orchestrator may be, when the config sent to it carries a credential. + +On the kafka_connect path, and on the DEFAULT ``bring-your-own`` / ``managed`` +modes of debezium and airbyte, forge-cli never connects to the source database. +It resolves the contract's ``connection.secretRef``, puts the plaintext into the +connector config (``database.password``, ``connection.password``) and POSTs the +whole config to ``properties..deployment.server_url`` -- a URL the +*contract* names. Nothing constrained that host, so a contract could have the +credential delivered anywhere on the internet, needing no allowlist entry and no +operator mistake. A loopback ``connection.host`` even made it look local. + +The policy, chosen by the founder on 2026-10-27 after the options were laid out: + +* A **public** host must be named in :data:`ALLOWLIST_ENV`. +* A **loopback or private** host needs no entry. An orchestrator on the cluster + network (``kafka-connect.internal:8083``, a ``docker compose`` service name) + is the ordinary deployment, and a credential that stays inside the + organisation is not the leak this closes. Refusing it would have broken every + working install on upgrade for no gain. +* **Link-local and unspecified are refused outright**, allowlist or not. That is + where cloud instance metadata lives (169.254.169.254), and it is never a + legitimate orchestrator. +* A host that does **not resolve** passes the construction-time check and is + judged again on every request. Refusing it outright was the first version and + it was wrong: ``kafka-connect.internal`` does not resolve on a laptop, so a + dry run against the ordinary cluster deployment failed with a security + message when the truth was "that host is not here" -- the very confusion + ``_net._hostname_is_link_local`` was written to avoid for ``docker compose`` + names. Nothing can be sent to a name that does not resolve + (``safe_http.resolve_and_validate`` raises before the request leaves), so + allowing it discloses nothing; :func:`request_guard` then classifies the host + for real once it does resolve, which closes the resolve-later race rather + than guessing at check time. + +This gate's polarity is the inverse of the SSRF gates in the tree -- it allows +internal and questions public -- so ``unresolved`` cannot simply be folded into +"not public" the way :func:`fluid_build._net._hostname_is_private` folds it. +:func:`fluid_build._net.hostname_scope` keeps the three apart for exactly this. + +Deliberately NOT modelled on ``FLUID_WEBHOOK_HOST_ALLOWLIST``, +``FLUID_FEDERATION_HOST_ALLOWLIST`` or ``FLUID_COMMAND_CENTER_HOST_ALLOWLIST``, +despite the matching name. Those are SSRF allowlists: unset, they *permit* any +public host and refuse private ones, and the Command Center gate still lets a +credential reach any public ``https`` endpoint unlisted. Mirroring them would +have reproduced the hole. The deny-unless-named shape comes from +``_secret_policy.py``, which answers the same question -- may this credential go +where the contract says? -- for ``connection.secretRef``. + +:data:`ALLOWLIST_ENV` is operator-only, registered in +``credentials/protected_env.py``: ``fluid apply`` hydrates the environment from +``.env``, ``.env.`` and ``FLUID_SECRETS_FILE``, which are committed beside +the contract, so without that a contract repository could grant itself the +allowlist meant to constrain it. + +**The entry grammar matches the three siblings** (comma-separated host suffixes, +``connect.example.com`` or ``example.com`` to cover its subdomains) because an +operator who has configured one of those should not have to learn a second +spelling. A port is never part of a host, so an entry carrying one never +matches; the refusal says so. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any, Callable, List, Mapping, Optional +from urllib.parse import urlsplit + +from fluid_build._errors import FluidUserError, doc_url +from fluid_build._net import _hostname_is_link_local, hostname_scope + +#: Operator-only. Comma-separated host suffixes. +ALLOWLIST_ENV = "FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST" + +_ALLOWED_SCHEMES = ("http", "https") + + +@dataclass +class DeploymentHostError(FluidUserError): + """An orchestrator URL may not receive a credential-bearing config. + + Raised where each runner reads ``deployment.server_url`` -- before the + connector config is built, and building it is what resolves the contract's + ``secretRef`` -- so a refused deployment never fetches the password from + its secret store. The REST client constructors check again, as the + chokepoint for any other caller. The client check ALONE was not enough: + the runners construct the client after the config, so the first version of + this promised an ordering it did not have. + """ + + code: str = "DeploymentHostError" + + +def allowlist(environ: Optional[Mapping[str, str]] = None) -> List[str]: + """The operator's host suffixes. + + 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 [s.strip() for s in (env.get(ALLOWLIST_ENV, "") or "").split(",") if s.strip()] + + +def _on_allowlist(host: str, suffixes: List[str]) -> bool: + """Exact match or dotted suffix, as the three sibling allowlists do.""" + lowered = host.strip().lower().rstrip(".") + for suffix in suffixes: + candidate = suffix.strip().lower().rstrip(".") + if not candidate: + continue + if lowered == candidate or lowered.endswith("." + candidate): + return True + return False + + +def host_refusal( + url: Optional[str], + *, + engine: str, + environ: Optional[Mapping[str, str]] = None, + operator_supplied: bool = False, +) -> Optional[str]: + """Why ``url`` may not receive a credential-bearing config, or ``None``. + + Split out from :func:`refuse_unless_host_allowed` so a caller that wants to + report rather than raise -- ``fluid doctor``, a dry run -- asks the same + question and cannot drift from the answer. + + ``operator_supplied`` says the URL came from a command-line flag or the + environment rather than from a contract. That is the whole threat model + here: the policy exists because a CONTRACT could choose where a credential + goes. An operator who types ``--server-url`` has chosen it themselves, and + refusing that would block a legitimate action while telling them to set an + allowlist variable to permit what they just asked for -- which is what + ``fluid import airbyte --server-url https://...`` did when this check was + first written. The scheme and metadata-address rules still apply, because + neither is ever what an operator meant.""" + if not url or not str(url).strip(): + return "no deployment.server_url is configured" + parts = urlsplit(str(url).strip()) + scheme = (parts.scheme or "").lower() + if scheme not in _ALLOWED_SCHEMES: + return ( + f"its scheme is {scheme or 'missing'}; the {engine} orchestrator must be an " + "http:// or https:// URL" + ) + try: + host = parts.hostname + except ValueError: + host = None + if not host: + return "it has no host" + # First, and not overridable: this is where instance metadata lives, and no + # orchestrator is ever legitimately there. + if _hostname_is_link_local(host): + return ( + f"host {host!r} is link-local or unspecified, where cloud instance metadata " + "lives; no orchestrator is there, and this is refused even with an allowlist " + "entry" + ) + if operator_supplied or _on_allowlist(host, allowlist(environ)): + return None + scope = hostname_scope(host) + if scope == "non_public": + return None + if scope == "mixed": + return ( + f"host {host!r} resolves to both public and non-public addresses, and the " + "connection is pinned to a public one. One private record beside a public " + "one is how an unnamed host is made to look internal, so a mixed answer " + "needs an allowlist entry like any other public host" + ) + if scope == "unresolved": + # Not this gate's refusal to make. A name that does not resolve cannot + # receive anything, and refusing it here turns "that host is not + # running" into a security error. Note what this does NOT buy: both + # clients pass the URL to safe_httpx_client, whose resolve_and_validate + # raises UnsafeURLError("cannot resolve hostname") at construction, so + # an unresolvable host is still refused -- by the transport layer, with + # an accurate message. This branch matters for a caller that reaches the + # policy without that layer, and for the resolve-later race that + # request_guard closes. + return None + return ( + f"host {host!r} is public, and the config posted to it carries the source " + "credential in clear text" + ) + + +def refuse_unless_host_allowed( + url: Optional[str], + *, + engine: str, + environ: Optional[Mapping[str, str]] = None, + operator_supplied: bool = False, +) -> str: + """``url``, or raise :class:`DeploymentHostError` saying why not. + + The message depends on who chose the URL. Blaming a contract when the + operator typed ``--server-url`` sends them looking for a contract field + they never set, and offering the allowlist as the remedy is wrong when the + refusal is a metadata address no entry can permit. + """ + refusal = host_refusal(url, engine=engine, environ=environ, operator_supplied=operator_supplied) + if refusal is None: + return str(url).strip() + if operator_supplied: + raise DeploymentHostError( + what=f"the {engine} endpoint given on the command line cannot be used", + why=f"Refused because {refusal}.", + fix=( + f"Point it at the {engine} API itself. A link-local address is instance " + "metadata, never a service endpoint, and is refused however it is " + "configured." + ), + doc=doc_url(), + ) + raise DeploymentHostError( + what=f"the {engine} orchestrator may not receive this product's credential", + why=( + f"forge-cli hands the whole connector config -- the resolved source password " + f"included -- to properties.{engine}.deployment.server_url, which the contract " + f"names. Refused because {refusal}." + ), + fix=( + f"If that orchestrator is yours, name its host in the OPERATOR environment (a " + f"shell or CI variable, not a project .env, which is ignored for this " + f'setting): {ALLOWLIST_ENV}="connect.example.com", or the parent domain to ' + "cover its subdomains. A host written with a port never matches, so leave " + "the port off. An orchestrator on this machine or on a private network needs " + "no entry." + ), + doc=doc_url(), + ) + + +def request_guard( + url: Optional[str], *, engine: str, operator_supplied: bool = False +) -> Callable[[Any], None]: + """An httpx request hook that re-applies the policy to ``url``'s host. + + The authoritative check. :func:`host_refusal` runs once at construction to + fail fast, but its verdict on a host that did not resolve yet is "allow", + so the policy has to be asked again when the request actually goes out -- + by then the name resolves or the request fails on its own. + + It closes over the URL given at construction rather than reading + ``request.url``, because ``safe_httpx_client`` inserts its DNS-pinning hook + ahead of this one and that hook rewrites the URL to the pinned IP. The + hostname is what an allowlist entry names, so the original is what must be + judged. + """ + + def _guard(request: Any) -> None: # pragma: no cover - exercised via the clients + refuse_unless_host_allowed(url, engine=engine, operator_supplied=operator_supplied) + + return _guard diff --git a/fluid_build/build_runners/airbyte/runner.py b/fluid_build/build_runners/airbyte/runner.py index 21803205..d93ccf44 100644 --- a/fluid_build/build_runners/airbyte/runner.py +++ b/fluid_build/build_runners/airbyte/runner.py @@ -42,6 +42,9 @@ ) from fluid_build.api.schema import SchemaFingerprint from fluid_build.api.security import ImageSignatureVerifier +from fluid_build.build_runners._deployment_host_policy import ( + refuse_unless_host_allowed, +) from .._acquisition_common import ( adapt_source_config, @@ -85,6 +88,7 @@ def __init__( *, api_token: Optional[str] = None, timeout_seconds: int = 30, + operator_supplied: bool = False, ): # SSRF guard — Airbyte server_url comes from contract.fluid.yaml # and carries a Bearer token in headers. allow_private=True for @@ -92,8 +96,22 @@ def __init__( # DNS-pins. follow_redirects=False (httpx already strips # Authorization cross-host, but a same-host redirect to an # alternate path would still receive the token). + # + # Where the config may go — see _deployment_host_policy. Airbyte's + # source config carries the resolved source password and this client + # carries the operator's Airbyte token, so a public host has to be one + # the operator named. The runner checks this where it reads server_url, + # before the config (and so the secretRef) is resolved; this is the + # chokepoint for any other caller. + # Not the module-level ``refuse_unless_allowed`` from _secret_policy: + # that polices which VARIABLE may be read for a given host, this one + # which HOST may receive the result. + from fluid_build.build_runners._deployment_host_policy import request_guard from fluid_build.util.safe_http import safe_httpx_client + server_url = refuse_unless_host_allowed( + server_url, engine="airbyte", operator_supplied=operator_supplied + ) headers: Dict[str, str] = {} if api_token: headers["Authorization"] = f"Bearer {api_token}" @@ -102,6 +120,11 @@ def __init__( headers=headers, timeout=float(timeout_seconds), allow_private=True, + event_hooks={ + "request": [ + request_guard(server_url, engine="airbyte", operator_supplied=operator_supplied) + ] + }, ) def close(self) -> None: @@ -306,6 +329,13 @@ def _execute_rest_mode( server_url = deployment.get("server_url") if not server_url: return _failed(ctx, started_at, t_start, "airbyte REST mode requires deployment.server_url") + + # Before the config is built, because building it resolves the contract's + # secretRef: the client constructor is the chokepoint, but it runs AFTER + # that, so a refused deployment would otherwise still have fetched the + # password from Vault/KMS -- audit entries, API calls, plaintext in memory. + # The constructor keeps its own check as the backstop for any other caller. + refuse_unless_host_allowed(server_url, engine="airbyte") auth = deployment.get("auth", {}) or {} # auth.token wins; if absent, resolve auth.secretRef via the central # secret backends (env://, vault://, aws://, gcp://, azure://, file://). diff --git a/fluid_build/build_runners/debezium/runner.py b/fluid_build/build_runners/debezium/runner.py index 451a9882..4f24ad57 100644 --- a/fluid_build/build_runners/debezium/runner.py +++ b/fluid_build/build_runners/debezium/runner.py @@ -46,6 +46,9 @@ StreamResult, ) from fluid_build.api.schema import SchemaFingerprint +from fluid_build.build_runners._deployment_host_policy import ( + refuse_unless_host_allowed, +) from .._acquisition_common import ( extract_source_schemas, @@ -353,6 +356,13 @@ def _execute_kafka_connect( t_start, "debezium kafka-connect mode requires deployment.server_url", ) + + # Before the config is built, because building it resolves the contract's + # secretRef: the client constructor is the chokepoint, but it runs AFTER + # that, so a refused deployment would otherwise still have fetched the + # password from Vault/KMS -- audit entries, API calls, plaintext in memory. + # The constructor keeps its own check as the backstop for any other caller. + refuse_unless_host_allowed(server_url, engine="debezium") try: connector_class = resolve_debezium_class( ctx.source.kind, override=dbz_props.get("connector_class") @@ -387,7 +397,7 @@ def _execute_kafka_connect( ) if late_arrival_policy.get("enabled"): config.update(late_arrival_policy["connector_config"]) - client = KafkaConnectRestClient(server_url) + client = KafkaConnectRestClient(server_url, engine="debezium") try: existing = client.get_connector(connector_name) if existing is None: diff --git a/fluid_build/build_runners/kafka_connect/runner.py b/fluid_build/build_runners/kafka_connect/runner.py index 43f88316..fee82975 100644 --- a/fluid_build/build_runners/kafka_connect/runner.py +++ b/fluid_build/build_runners/kafka_connect/runner.py @@ -42,6 +42,10 @@ StreamResult, ) from fluid_build.api.schema import SchemaFingerprint +from fluid_build.build_runners._deployment_host_policy import ( + DeploymentHostError, + refuse_unless_host_allowed, +) from .._acquisition_common import ( extract_source_schemas, @@ -117,19 +121,40 @@ def _find_iceberg_expose_binding( class KafkaConnectRestClient: """Minimal REST client for the Kafka Connect API.""" - def __init__(self, base_url: str, *, timeout_seconds: int = 30): - # SSRF guard — Kafka Connect ``server_url`` comes from - # contract.fluid.yaml, which may be operator-supplied. Route - # through safe_httpx_client so a poisoned URL pointing at IMDS - # / RFC1918 cannot exfil connector configs (which carry DB - # creds). allow_private=True because Kafka Connect endpoints - # commonly live on private cluster networks. + def __init__( + self, + base_url: str, + *, + timeout_seconds: int = 30, + engine: str = "kafka-connect", + operator_supplied: bool = False, + ): + # Where the config may go. The connector config POSTed through this + # client carries the resolved source password, and ``server_url`` is + # named by contract.fluid.yaml, so a public host has to be one the + # OPERATOR named. Checked here, at construction, because that is before + # the runner resolves the secretRef: a refused deployment never reads + # the credential. ``allow_private=True`` below is what left this open — + # it is the right setting for a cluster-internal orchestrator and no + # constraint at all on a public one. + # + # ``engine`` only names the caller in the message; debezium builds this + # same client for its kafka-connect mode. + from fluid_build.build_runners._deployment_host_policy import request_guard from fluid_build.util.safe_http import safe_httpx_client + base_url = refuse_unless_host_allowed( + base_url, engine=engine, operator_supplied=operator_supplied + ) self._client = safe_httpx_client( base_url=base_url, timeout=float(timeout_seconds), allow_private=True, + event_hooks={ + "request": [ + request_guard(base_url, engine=engine, operator_supplied=operator_supplied) + ] + }, ) def close(self) -> None: @@ -240,6 +265,11 @@ def _poll_connector_health( if state in ("RUNNING", "FAILED", "PAUSED"): ok = state == "RUNNING" and not any(t.get("state") == "FAILED" for t in tasks) return state, ok, traces + except DeploymentHostError: + # Not transient, and not something polling can resolve: the guard + # refused the host this client dials. Retrying to the deadline + # would swallow it and report UNKNOWN instead. + raise except Exception as exc: # noqa: BLE001 — eventual consistency LOG.debug("kafka_connect.status_poll.transient name=%s err=%s", name, exc) time.sleep(interval) @@ -273,6 +303,13 @@ def _execute(ctx: RunContext, runner: KafkaConnectRunner) -> RunResult: if not server_url: return _failed(ctx, started_at, t_start, "kafka-connect requires deployment.server_url") + # Before the config is built, because building it resolves the contract's + # secretRef: the client constructor is the chokepoint, but it runs AFTER + # that, so a refused deployment would otherwise still have fetched the + # password from Vault/KMS -- audit entries, API calls, plaintext in memory. + # The constructor keeps its own check as the backstop for any other caller. + refuse_unless_host_allowed(server_url, engine="kafka-connect") + try: connector_class = resolve_source_connector( ctx.source.kind, override=kc_props.get("connector_class") @@ -487,6 +524,11 @@ def _execute(ctx: RunContext, runner: KafkaConnectRunner) -> RunResult: "late_arrival_side_output_table": late_arrival_policy.get("side_output_table"), }, ) + except DeploymentHostError: + # Let the typed refusal out whole: _failed(str(exc)) keeps only `what` + # and drops the `why` and the `fix` naming the allowlist variable, + # which is the whole value of the message. + raise except Exception as exc: # noqa: BLE001 LOG.error("kafka_connect.failed err=%s", exc, exc_info=True) return _failed(ctx, started_at, t_start, str(exc)) diff --git a/fluid_build/cli/import_workflow/airbyte.py b/fluid_build/cli/import_workflow/airbyte.py index 36189f0f..27c64e04 100644 --- a/fluid_build/cli/import_workflow/airbyte.py +++ b/fluid_build/cli/import_workflow/airbyte.py @@ -92,7 +92,17 @@ def import_to_contract( from fluid_build.build_runners.airbyte.runner import AirbyteRestClient token = opts.get("api_token", self.api_token) - client = AirbyteRestClient(url, api_token=token, timeout_seconds=self.timeout_seconds) + # The operator chose this URL (``--server-url`` or + # FLUID_IMPORT_AIRBYTE_URL), not a contract, so the + # deployment-host policy does not second-guess it. Reading an + # Airbyte workspace the operator named is the point of the + # command; the scheme and metadata-address rules still apply. + client = AirbyteRestClient( + url, + api_token=token, + timeout_seconds=self.timeout_seconds, + operator_supplied=True, + ) report = ImportReport() try: sources = client.list_sources(workspace_id=source) diff --git a/fluid_build/credentials/protected_env.py b/fluid_build/credentials/protected_env.py index b81d0916..15af94a0 100644 --- a/fluid_build/credentials/protected_env.py +++ b/fluid_build/credentials/protected_env.py @@ -265,6 +265,7 @@ "FLUID_MCP_RATE_LIMIT", "FLUID_MCP_RATE_WINDOW_SECONDS", "FLUID_MCP_MAX_CONCURRENCY", + "FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST", "FLUID_COMMAND_CENTER_HOST_ALLOWLIST", "FLUID_FEDERATION_HOST_ALLOWLIST", "FLUID_FEDERATION_SECRET_REF_ALLOWLIST", diff --git a/fluid_build/util/safe_http.py b/fluid_build/util/safe_http.py index d7d4cb35..3bb3a58a 100644 --- a/fluid_build/util/safe_http.py +++ b/fluid_build/util/safe_http.py @@ -60,6 +60,8 @@ from typing import Any, Optional, Tuple from urllib.parse import urlparse +from fluid_build import _net + LOG = logging.getLogger(__name__) DEFAULT_TIMEOUT = 30.0 # seconds @@ -70,60 +72,14 @@ class UnsafeURLError(ValueError): """Raised when a URL targets a non-public address or non-http(s) scheme.""" -# CIDR ranges that ``ipaddress.is_private`` does NOT flag but which are -# nonetheless non-routable / not safe to fetch from. Borrowed from -# requests-hardened (Saleor, BSD-3). -EXTRA_BLOCKED_NETWORKS = tuple( - ipaddress.ip_network(cidr) - for cidr in ( - "100.64.0.0/10", # RFC 6598 — Carrier-Grade NAT - "192.0.0.0/24", # RFC 6890 — IETF Protocol Assignments (CPython #113171 gap) - "192.0.2.0/24", # RFC 5737 — TEST-NET-1 - "192.88.99.0/24", # RFC 7526 — 6to4 relay anycast (deprecated) - "198.18.0.0/15", # RFC 2544 — benchmarking - "198.51.100.0/24", # RFC 5737 — TEST-NET-2 - "203.0.113.0/24", # RFC 5737 — TEST-NET-3 - "240.0.0.0/4", # RFC 1112 — class E reserved - "2001:20::/28", # RFC 7343 — ORCHIDv2 - "2002::/16", # RFC 3056 — 6to4 - "5f00::/16", # RFC 9602 — IPv6 Segment Routing - "64:ff9b::/96", # RFC 6052 — Well-Known NAT64 - "64:ff9b:1::/48", # RFC 8215 — Local-Use NAT64 - ) -) - - -def is_public_address(addr: str) -> bool: - """True iff ``addr`` is a routable public IP (v4 or v6). - - Handles three traps ``ipaddress`` alone does not: - * IPv4-mapped IPv6 (``::ffff:a.b.c.d``) — on Python <3.12 the - v6 form does NOT recurse ``is_private``/``is_loopback`` into - the embedded v4 (gh-87105 / bpo-44904 fixed only in 3.12). - * Zone-id-bearing v6 (``fe80::1%eth0``) — strip the scope. - * Extended CIDR deny-list (:data:`EXTRA_BLOCKED_NETWORKS`). - """ - if "%" in addr: - addr = addr.split("%", 1)[0] - try: - ip = ipaddress.ip_address(addr) - except ValueError: - return False - if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: - ip = ip.ipv4_mapped - if ( - ip.is_private - or ip.is_loopback - or ip.is_link_local - or ip.is_multicast - or ip.is_reserved - or ip.is_unspecified - ): - return False - for network in EXTRA_BLOCKED_NETWORKS: - if ip.version == network.version and ip in network: - return False - return True +# The address predicate and the extended deny-list now live in the tier-0 leaf +# ``fluid_build._net``, beside the broad SSRF gate that consults the same list. +# They were defined in both places and disagreed: ``_net._hostname_is_private`` +# judged on bare ``ipaddress`` flags, so a CGNAT address was "public" there and +# non-public here. Re-exported under the original names because both are part +# of this module's documented surface and ``__all__``. +EXTRA_BLOCKED_NETWORKS = _net.EXTRA_BLOCKED_NETWORKS +is_public_address = _net.is_public_address def resolve_and_validate(hostname: str, *, allow_private: bool = False) -> str: diff --git a/tests/build_runners/test_deployment_host_policy.py b/tests/build_runners/test_deployment_host_policy.py new file mode 100644 index 00000000..1d92dca2 --- /dev/null +++ b/tests/build_runners/test_deployment_host_policy.py @@ -0,0 +1,530 @@ +# 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 public orchestrator may not receive a product's credential unnamed. + +On the kafka-connect path, and on the DEFAULT ``bring-your-own`` / ``managed`` +modes of debezium and airbyte, forge-cli never connects to the source database: +it resolves the contract's ``secretRef``, puts the plaintext into the connector +config and POSTs the whole config to +``properties..deployment.server_url`` -- a URL the contract names. + +The policy: a PUBLIC host must be named in +``FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST``; loopback and private hosts need +no entry; link-local is refused outright; an operator-supplied URL is not +second-guessed. + +**DNS is faked throughout.** An earlier version of this file resolved +``example.com`` for its public cases and skipped when that failed, which meant +six of its eight DNS-dependent assertions could pass without testing anything, +and the comment claiming otherwise was wrong. Every host here is answered by +:func:`_fake_getaddrinfo`, so the suite is deterministic and offline, and the +mixed-answer case -- which no earlier test covered and which was a complete +bypass -- can be stated exactly. + +The three things this file exists to pin, each of which was broken when first +written and is mutation-checked: + +1. **A mixed DNS answer is not "internal".** One private record beside a public + one made ``hostname_scope`` say ``non_public``, the policy allow the host + unlisted, and ``safe_http`` pin the public address -- the exact leak the + policy exists to stop, with no operator error. +2. **The host is judged before the secret is resolved.** The client constructor + is the chokepoint, but the runners build the connector config first, and + that resolves the ``secretRef``. The gate therefore also runs where + ``server_url`` is read, and the test drives the real runner with the spy + patched where the runner looks the name up. +3. **The per-request guard really fires.** Asserting ``len(event_hooks) >= 2`` + passed for a no-op guard; this fires a request and asserts nothing is sent. +""" + +from __future__ import annotations + +import json +import pathlib +import socket +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import Any, Dict, Iterator, List, Tuple + +import pytest + +from fluid_build._net import _hostname_is_private, hostname_scope, is_public_address +from fluid_build.build_runners._deployment_host_policy import ( + ALLOWLIST_ENV, + DeploymentHostError, + host_refusal, + refuse_unless_host_allowed, + request_guard, +) + +#: Synthetic hosts and the addresses DNS will answer for them. Names, not live +#: domains, so no assertion here depends on the network or on who owns what. +FAKE_DNS: Dict[str, List[str]] = { + "orchestrator.public.test": ["93.184.216.34"], + "orchestrator.private.test": ["10.1.2.3"], + "orchestrator.loopback.test": ["127.0.0.1"], + "orchestrator.cgnat.test": ["100.100.100.100"], # Tailscale / RFC 6598 + "orchestrator.metadata.test": ["169.254.169.254"], + # The bypass: a decoy private record beside the attacker's public one. + "orchestrator.mixed.test": ["fd00::1", "93.184.216.34"], + "orchestrator.v6public.test": ["2606:2800:220:1:248:1893:25c8:1946"], + "orchestrator.mappedprivate.test": ["::ffff:10.1.2.3"], +} + + +_REAL_GETADDRINFO = socket.getaddrinfo + + +def _fake_getaddrinfo(host: str, *a: Any, **k: Any) -> List[Tuple[Any, ...]]: + """Answer from :data:`FAKE_DNS`; a NAME outside it does not resolve. + + A literal address is passed to the real resolver, which does not touch the + network for one, so ``127.0.0.1`` and ``169.254.169.254`` behave normally + and the loopback listener test still works. Only names are synthetic, which + is the part that would otherwise depend on DNS. + """ + import ipaddress as _ip + + try: + _ip.ip_address(str(host).strip("[]").split("%", 1)[0]) + except ValueError: + pass + else: + return _REAL_GETADDRINFO(host, *a, **k) + if host not in FAKE_DNS: + raise socket.gaierror(socket.EAI_NONAME, "Name or service not known") + out: List[Tuple[Any, ...]] = [] + for addr in FAKE_DNS[host]: + if ":" in addr: + out.append((socket.AF_INET6, socket.SOCK_STREAM, 6, "", (addr, 0, 0, 0))) + else: + out.append((socket.AF_INET, socket.SOCK_STREAM, 6, "", (addr, 0))) + return out + + +@pytest.fixture(autouse=True) +def _dns(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + """Every test in this file resolves through :data:`FAKE_DNS`. + + Patched on ``socket`` itself, which is what ``_net`` and ``safe_http`` both + call, so the policy and the transport layer agree on what a host is. + """ + monkeypatch.setattr(socket, "getaddrinfo", _fake_getaddrinfo) + monkeypatch.delenv(ALLOWLIST_ENV, raising=False) + yield + + +# ── the classifier the policy needed ──────────────────────────────────── + + +@pytest.mark.parametrize( + "host, scope", + [ + ("orchestrator.public.test", "public"), + ("orchestrator.v6public.test", "public"), + ("orchestrator.private.test", "non_public"), + ("orchestrator.loopback.test", "non_public"), + ("orchestrator.metadata.test", "non_public"), + ("orchestrator.cgnat.test", "non_public"), + ("orchestrator.mappedprivate.test", "non_public"), + ("orchestrator.mixed.test", "mixed"), + ("orchestrator.absent.test", "unresolved"), + ], +) +def test_hostname_scope_examines_every_address(host: str, scope: str) -> None: + assert hostname_scope(host) == scope + + +@pytest.mark.parametrize( + "host, private", + [ + ("orchestrator.public.test", False), + ("orchestrator.v6public.test", False), + ("orchestrator.private.test", True), + ("orchestrator.loopback.test", True), + ("orchestrator.metadata.test", True), + ("orchestrator.mixed.test", True), + ("orchestrator.absent.test", True), + ("orchestrator.cgnat.test", True), + ], +) +def test_the_broad_gate_keeps_its_fail_closed_direction(host: str, private: bool) -> None: + """``_hostname_is_private`` has five callers and must not loosen. + + Both directions are asserted, not just the True ones: a function that + always returned True would satisfy a one-sided test and silently break + every SSRF gate's ability to allow a legitimate public fetch. + + ``mixed`` and ``unresolved`` answer True here -- that is this gate's safe + direction, and the opposite of the deployment-host policy's, which is why + the classifier keeps the states apart. CGNAT answers True where it once + answered False: the extended deny-list is now consulted here too, which + only ever refuses more and matches what ``safe_http`` already did. + """ + assert _hostname_is_private(host) is private + + +def test_the_address_predicate_is_shared_not_copied() -> None: + """``safe_http`` re-exports the tier-0 definitions rather than owning them. + + They disagreed before: bare ``ipaddress`` flags here, the extended + deny-list there, so one CGNAT address was public to one and private to the + other. + """ + from fluid_build.util import safe_http + + assert safe_http.is_public_address is is_public_address + from fluid_build import _net + + assert safe_http.EXTRA_BLOCKED_NETWORKS is _net.EXTRA_BLOCKED_NETWORKS + + +# ── the policy ─────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "url, listed, allowed, because", + [ + ("http://orchestrator.loopback.test:8083", "", True, "loopback needs no entry"), + ("http://orchestrator.private.test:8083", "", True, "the ordinary cluster install"), + ("http://orchestrator.cgnat.test:8083", "", True, "Tailscale is an internal network"), + ("https://orchestrator.public.test", "", False, "a public host must be named"), + ("https://orchestrator.public.test", "orchestrator.public.test", True, "named exactly"), + ("https://orchestrator.public.test", "public.test", True, "dotted suffix, as siblings do"), + ("https://orchestrator.public.test", "other.test", False, "a different entry is not this"), + ( + "https://orchestrator.public.test", + "ublic.test", + False, + "a bare substring is not a suffix", + ), + ( + "https://orchestrator.public.test", + "orchestrator.public.test:443", + False, + "ports never match", + ), + ("https://ORCHESTRATOR.PUBLIC.TEST", "orchestrator.public.test", True, "case-insensitive"), + ("https://orchestrator.public.test.", "orchestrator.public.test", True, "trailing dot"), + ("https://orchestrator.mixed.test", "", False, "a decoy private record is still public"), + ("https://orchestrator.mixed.test", "mixed.test", True, "named, so the operator chose it"), + ("http://orchestrator.metadata.test", "", False, "instance metadata"), + ("http://orchestrator.metadata.test", "metadata.test", False, "metadata, even named"), + ("http://169.254.169.254", "169.254.169.254", False, "by address, even named"), + ("http://0.0.0.0", "", False, "unspecified routes to loopback on many stacks"), + ("file:///etc/passwd", "", False, "not an http(s) orchestrator"), + ("", "", False, "nothing configured"), + ("https://", "", False, "no host"), + ], +) +def test_the_policy_matrix(url: str, listed: str, allowed: bool, because: str) -> None: + refusal = host_refusal(url, engine="kafka-connect", environ={ALLOWLIST_ENV: listed}) + assert (refusal is None) is allowed, f"{because}: {url!r} listed={listed!r} -> {refusal!r}" + + +def test_a_mixed_answer_is_refused_for_what_it_is() -> None: + """The message has to name the real reason, or the reader cannot act. + + "is public" would be misleading for a host that does answer with a private + address; the operator needs to know a decoy record is what they are seeing. + """ + refusal = host_refusal("https://orchestrator.mixed.test", engine="airbyte", environ={}) + assert refusal is not None + assert "both public and non-public" in refusal, refusal + + +def test_an_unresolvable_host_is_not_this_gate_s_refusal() -> None: + """It is refused by the transport layer instead, with an accurate message. + + Refusing it here turned "that host is not running" into a security error + for ``kafka-connect.internal`` on a developer's machine. + """ + assert host_refusal("http://orchestrator.absent.test:8083", engine="kafka-connect") is None + + # The transport layer is what refuses it -- ``resolve_and_validate`` raises + # ``UnsafeURLError("cannot resolve hostname")`` before a request leaves. Not + # asserted here: ``tests/conftest.py::_disable_ssrf_guard_for_unit_tests`` + # no-ops ``assert_safe_url`` for every file except the two that exercise the + # guard for real, so an assertion here would pass or fail on the stub rather + # than on the guard. ``tests/util/test_safe_http.py`` owns that case. + + +def test_an_operator_chosen_url_is_not_second_guessed() -> None: + """A CLI flag or env var is the operator's choice, not a contract's.""" + assert host_refusal("https://orchestrator.public.test", engine="airbyte") is not None + assert ( + host_refusal("https://orchestrator.public.test", engine="airbyte", operator_supplied=True) + is None + ) + + +@pytest.mark.parametrize( + "url", + ["http://orchestrator.metadata.test", "http://0.0.0.0", "file:///etc/passwd"], +) +def test_what_an_operator_can_never_ask_for(url: str) -> None: + assert host_refusal(url, engine="airbyte", operator_supplied=True) is not None + + +def test_the_refusal_names_the_variable_the_host_and_the_port_trap() -> None: + with pytest.raises(DeploymentHostError) as excinfo: + refuse_unless_host_allowed("https://orchestrator.public.test", engine="debezium") + error = excinfo.value + rendered = " ".join([str(error), error.why or "", error.fix or ""]) + assert ALLOWLIST_ENV in rendered + assert "orchestrator.public.test" in rendered + assert "debezium" in rendered + assert "port" in rendered + assert error.code == "DeploymentHostError" + + +def test_the_operator_facing_refusal_does_not_blame_a_contract() -> None: + with pytest.raises(DeploymentHostError) as excinfo: + refuse_unless_host_allowed( + "http://orchestrator.metadata.test", engine="airbyte", operator_supplied=True + ) + error = excinfo.value + rendered = " ".join([str(error), error.why or "", error.fix or ""]) + assert "contract" not in rendered, rendered + assert ALLOWLIST_ENV not in rendered, rendered + assert "metadata" in rendered, rendered + + +def test_the_variable_is_operator_only() -> None: + """A project `.env` beside the contract must not grant the allowlist.""" + from fluid_build.credentials.protected_env import is_protected_env_key + + assert is_protected_env_key(ALLOWLIST_ENV) + + +def test_the_allowlist_is_read_from_the_process_environment( + monkeypatch: pytest.MonkeyPatch, +) -> None: + assert host_refusal("https://orchestrator.public.test", engine="airbyte") is not None + monkeypatch.setenv(ALLOWLIST_ENV, "orchestrator.public.test") + assert host_refusal("https://orchestrator.public.test", engine="airbyte") is None + + +# ── the clients ────────────────────────────────────────────────────────── + + +def test_both_rest_clients_refuse_a_public_unnamed_orchestrator( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from fluid_build.build_runners.airbyte.runner import AirbyteRestClient + from fluid_build.build_runners.kafka_connect.runner import KafkaConnectRestClient + + with pytest.raises(DeploymentHostError): + KafkaConnectRestClient("https://orchestrator.public.test") + with pytest.raises(DeploymentHostError): + AirbyteRestClient("https://orchestrator.public.test", api_token="t") + + monkeypatch.setenv(ALLOWLIST_ENV, "orchestrator.public.test") + KafkaConnectRestClient("https://orchestrator.public.test").close() + AirbyteRestClient("https://orchestrator.public.test", api_token="t").close() + + +def test_the_engine_named_in_the_refusal_is_the_caller_s() -> None: + """debezium shares the kafka-connect client; the message must not say so.""" + from fluid_build.build_runners.kafka_connect.runner import KafkaConnectRestClient + + with pytest.raises(DeploymentHostError) as excinfo: + KafkaConnectRestClient("https://orchestrator.public.test", engine="debezium") + assert "debezium" in (excinfo.value.why or "") + + +# ── the per-request guard, fired for real ──────────────────────────────── + + +def test_the_guard_refuses_a_host_that_only_resolves_later( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Construction allows an unresolved host; the request must not go out. + + The previous version of this test asserted ``len(event_hooks) >= 2``, which + a no-op guard satisfies. This builds the client while the name does not + resolve, makes it resolve to a public address, fires a real request, and + asserts both the refusal and that the listener received nothing. + """ + from fluid_build.build_runners.kafka_connect.runner import KafkaConnectRestClient + + received: List[str] = [] + + class _Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - stdlib spelling + received.append(self.path) + self.send_response(200) + self.send_header("content-type", "application/json") + self.end_headers() + self.wfile.write(b"[]") + + def log_message(self, *_a: Any) -> None: + pass + + server = HTTPServer(("127.0.0.1", 0), _Handler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + # Unresolved at construction: allowed, and no request has gone out. + client = KafkaConnectRestClient("http://orchestrator.late.test:%d" % port) + try: + # Now it resolves -- to the listener, which is loopback, so prove + # the guard judges the NAME's scope rather than the socket: make + # the name answer with a public address while the connection still + # lands on the listener. + FAKE_DNS["orchestrator.late.test"] = ["93.184.216.34"] + with pytest.raises(DeploymentHostError): + client.list_connectors() + assert received == [], f"the request must not reach the orchestrator, got {received}" + finally: + FAKE_DNS.pop("orchestrator.late.test", None) + client.close() + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_the_guard_judges_the_original_host_not_the_pinned_ip() -> None: + """The SSRF hook rewrites the URL to an IP before this one runs.""" + + class _Explodes: + @property + def url(self) -> str: + raise AssertionError("the guard must not read the rewritten request URL") + + import os + + os.environ[ALLOWLIST_ENV] = "orchestrator.public.test" + try: + request_guard("https://orchestrator.public.test", engine="airbyte")(_Explodes()) + finally: + del os.environ[ALLOWLIST_ENV] + + +@pytest.mark.parametrize( + "module", + [ + "fluid_build.build_runners.kafka_connect.runner", + "fluid_build.build_runners.airbyte.runner", + ], +) +def test_every_client_call_uses_a_relative_path(module: str) -> None: + """An absolute URL would evade a guard that closes over the base URL.""" + import importlib + import re + + source = pathlib.Path(importlib.import_module(module).__file__).read_text(encoding="utf-8") + calls = re.findall(r"self\._client\.(?:get|post|put|delete|request)\(\s*(.)", source) + assert calls, f"no client calls found in {module}; has the client moved?" + assert not re.search( + r"self\._client\.(?:get|post|put|delete|request)\(\s*f?\"https?://", source + ), f"{module}: a client call passes an absolute URL, which the guard cannot police" + + +# ── the ordering guarantee, driven through the real runner ─────────────── + + +def test_a_refused_deployment_never_resolves_the_secret( + monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path +) -> None: + """The claim the docs make, tested where it can actually fail. + + The client constructor is the chokepoint, but the runner builds the + connector config first and THAT resolves the ``secretRef`` -- so the gate + also runs where ``server_url`` is read. Two things the earlier version of + this test got wrong: it constructed the client directly instead of driving + the runner, and it patched ``_acquisition_common.resolve_connection_secrets`` + while the runner binds that name in its own module namespace, so the patch + could never have been reached. + """ + import fluid_build.build_runners.kafka_connect.runner as kc + from tests.build_runners.test_kafka_connect_full_matrix import _base_contract + + def _never(*_a: Any, **_k: Any) -> Any: + raise AssertionError("a secret was resolved for a refused orchestrator") + + monkeypatch.setattr(kc, "resolve_connection_secrets", _never) + + contract = _base_contract( + source={ + "kind": "postgres", + "connection": { + "host": "db.internal", + "port": 5432, + "database": "x", + "user": "u", + "secretRef": "env://DB_SECRET", # pragma: allowlist secret + }, + "mode": "full_refresh", + "streams": ["public.orders"], + }, + kc_props={"deployment": {"server_url": "https://orchestrator.public.test"}}, + ) + + with pytest.raises(DeploymentHostError): + kc.execute_kafka_connect_build(contract["builds"][0], contract, tmp_path, dry_run=False) + + +def test_a_loopback_orchestrator_still_receives_the_connector_config() -> None: + """The policy must not break the local path it exempts. + + Also the honest demonstration of the defect: the POSTed body carries the + resolved password. It goes to a listener on this machine. + """ + from fluid_build.build_runners.kafka_connect.runner import KafkaConnectRestClient + + received: List[Tuple[str, str]] = [] + + class _Captor(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 - stdlib spelling + n = int(self.headers.get("content-length") or 0) + received.append((self.path, self.rfile.read(n).decode() if n else "")) + self.send_response(201) + self.send_header("content-type", "application/json") + self.end_headers() + self.wfile.write(b'{"name": "probe"}') + + def log_message(self, *_a: Any) -> None: + pass + + server = HTTPServer(("127.0.0.1", 0), _Captor) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + client = KafkaConnectRestClient(f"http://127.0.0.1:{server.server_address[1]}") + try: + client.create_connector( + "probe", + { + "connector.class": "io.debezium.connector.postgresql.PostgresConnector", + "database.hostname": "db.internal", + "database.password": CONTRACT_PASSWORD, + }, + ) + finally: + client.close() + finally: + server.shutdown() + thread.join(timeout=5) + + assert received, "the loopback deployment must still be reachable" + path, body = received[0] + assert path == "/connectors" + assert json.loads(body)["config"]["database.password"] == CONTRACT_PASSWORD + + +#: Stands in for a password the runner resolved from the contract. +CONTRACT_PASSWORD = "resolved-from-the-contract" # pragma: allowlist secret diff --git a/tests/cli/import_workflow/test_airbyte_server_url_required.py b/tests/cli/import_workflow/test_airbyte_server_url_required.py index b4f41d8e..a6defd1c 100644 --- a/tests/cli/import_workflow/test_airbyte_server_url_required.py +++ b/tests/cli/import_workflow/test_airbyte_server_url_required.py @@ -148,6 +148,7 @@ class _FakeRestClient: """ last_url: Optional[str] = None + last_operator_supplied: Optional[bool] = None def __init__( self, @@ -155,8 +156,10 @@ def __init__( *, api_token: Optional[str] = None, timeout_seconds: int = 30, + operator_supplied: bool = False, ) -> None: type(self).last_url = server_url + type(self).last_operator_supplied = operator_supplied self.closed = False def list_sources(self, workspace_id: str) -> List[Dict[str, Any]]: @@ -298,6 +301,12 @@ def test_flag_reaches_the_real_client(self, fake_client: type) -> None: source=WORKSPACE, ) assert fake_client.last_url == FLAG_URL + # The URL is the OPERATOR's, typed on the command line, so the + # deployment-host policy must not second-guess it: without this flag, + # `--server-url https://airbyte.example.com` was refused and the + # operator was told to set an allowlist variable to permit what they + # had just asked for. + assert fake_client.last_operator_supplied is True def test_env_var_reaches_the_real_client( self, fake_client: type, monkeypatch: pytest.MonkeyPatch From c14a25d0ee1431860bd91823b1f17ede3e1073a3 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 27 Oct 2026 23:15:02 +0100 Subject: [PATCH 2/2] test(build_runners): one import style for the kafka_connect runner The ordering test reached the runner module through a dotted `import ... as` while every other test in the file used `from ... import`, which the quality bot flagged on #758. The from-import of the module object keeps the monkeypatch target (the runner's own namespace, which is the whole point of that test) without a `sys.modules` lookup. --- tests/build_runners/test_deployment_host_policy.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/build_runners/test_deployment_host_policy.py b/tests/build_runners/test_deployment_host_policy.py index 1d92dca2..bc9bfe20 100644 --- a/tests/build_runners/test_deployment_host_policy.py +++ b/tests/build_runners/test_deployment_host_policy.py @@ -450,7 +450,7 @@ def test_a_refused_deployment_never_resolves_the_secret( while the runner binds that name in its own module namespace, so the patch could never have been reached. """ - import fluid_build.build_runners.kafka_connect.runner as kc + from fluid_build.build_runners.kafka_connect import runner as kc from tests.build_runners.test_kafka_connect_full_matrix import _base_contract def _never(*_a: Any, **_k: Any) -> Any: