From 12057f78a95d5ff5309b91cc9303788c0ff1e1c2 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 4 Nov 2026 05:35:58 +0100 Subject: [PATCH 1/6] security(acquisition): the orchestrator guard judges the pinned address, not a second lookup (#814) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * security(acquisition): the orchestrator guard judges the pinned address, not a second lookup request_guard (#758) is the per-request check that keeps a contract-named deployment.server_url from receiving the connector config, resolved source password included, unless the host is private, loopback or listed in FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST. It re-ran the name check, which resolved the hostname again instead of judging the IP safe_httpx_client's pin hook had already written into request.url. The two lookups can disagree: - DNS rebinding: the pin lookup answers public, the guard's answers private, so an unlisted public address received the config. A listed name could be pointed at instance metadata the same way. - IDNA: socket.getaddrinfo encodes faß.de as IDNA2003 fass.de while httpx resolves IDNA2008 xn--fa-hia.de. Both reproduced end to end against the real KafkaConnectRestClient and pin hook (getaddrinfo faked, transport recorded): on main the request reached the transport for 93.184.216.34; now it is refused before it leaves. The guard now runs the name check as before, then pinned_verdict on request.url.host: link-local is refused always, the URL's name decides the allowlist, the pinned address decides the scope, and the refusal names both ("host 'x', connecting to '93.184.216.34', is public"). VerdictCode, HostVerdict (with pinned), host_verdict and pinned_verdict are copied verbatim from the managed Flink branch (5466a507), where catalog_request_guard already works this way, so the #784 sync merges these hunks cleanly. host_verdict returns no-host for http://[::1 instead of raising ValueError, so a malformed server_url is refused, not a traceback. The unit suite replaces the pin hook with a no-op, so the guard sees the mock orchestrator's name, which does not resolve and is refused. The kafka_connect_mock and airbyte_mock fixtures now list their host, as an operator would for an orchestrator of theirs. The same dial-time shape: richardwooding/ssrfguard's Guard.Control and the datamodel-code-generator 0.63.0 fix for CVE-2026-55391. * chore(types): put the acquisition egress policies under mypy --strict _deployment_host_policy and _secret_policy decide which host a contract-named URL may reach and which variable may be sent there. A silent Any is a silent allow, and strict equality is what makes a misspelt VerdictCode fail CI instead of falling through to the "public" wording. Same entries and comment as the managed Flink branch (5466a507), minus fluid_build.streaming.policy, which is not on main. * docs(changelog): the orchestrator guard judges the pinned address Security entry for the request_guard fix. * fix(acquisition): say the request has no address when nothing was pinned A pinned verdict with no address is not a URL without a host, so the refusal no longer reads "it has no host" there. Review finding. * test(acquisition): drive the guard through the real pin hook The unit suite replaces safe_http's pin hook with a no-op, and the guard tests hand it a stand-in request, so nothing checked the ordering the fix relies on: the pin hook runs first and the guard reads the IP it wrote. Both reviewers showed that moving the pin hook after the guard left the suite green while the faß.de case reached a public address again. The new file is on conftest's real-guard list. It drives the real KafkaConnectRestClient and AirbyteRestClient, with getaddrinfo faked and the transport recording instead of sending. A rebinding name and faß.de are refused with nothing sent, and a private orchestrator is reached at its pinned address. Moving the pin hook last, or restoring the old guard, fails 3 of its 4 tests. --- CHANGELOG.md | 16 ++ .../build_runners/_deployment_host_policy.py | 198 ++++++++++++++---- pyproject.toml | 6 + tests/_infrastructure/respx_fixtures.py | 20 +- .../test_deployment_host_policy.py | 91 ++++++-- ...st_deployment_host_policy_real_pin_hook.py | 123 +++++++++++ tests/conftest.py | 1 + 7 files changed, 401 insertions(+), 54 deletions(-) create mode 100644 tests/build_runners/test_deployment_host_policy_real_pin_hook.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6deb6e95f..ec3befd4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -546,6 +546,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 into `main.tf.json`. The module's allowlists now end in `\Z`, so such a name is refused with `Invalid SQL identifier: 'spectrum_events\n'`. That includes a name that reaches it with a trailing newline from an environment variable or a file; strip the value where it is set. + +- **The acquisition orchestrator check judged a second DNS answer, not the address the config + went to.** For kafka_connect, debezium and airbyte, the per-request guard that keeps a + contract-named `deployment.server_url` from receiving the connector config, resolved source + password included, unless its host is private, loopback or listed in + `FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST`, resolved the hostname again instead of looking at + the IP `safe_httpx_client` had pinned the connection to. A name that answered the two lookups + differently (DNS rebinding), or a non-ASCII name that Python's resolver spells as IDNA2003 and + httpx as IDNA2008 (`faß.de` is `fass.de` to one and `xn--fa-hia.de` to the other), passed as + private while the config went to an unlisted public address, and a listed name could be sent + to instance metadata the same way. The guard now judges the pinned address. The name still + decides the allowlist, the address decides public or private, a link-local address is refused + even for a listed name, and the refusal names both: `host 'connect.example.com', connecting to + '93.184.216.34', is public`. A URL that cannot be parsed, such as `http://[::1`, is refused with + "it has no host" where it raised `ValueError`. + ### Upgrade notes - A contract with a column typed `array`, `struct`, `map`, `object`, `record` or `row` (or diff --git a/fluid_build/build_runners/_deployment_host_policy.py b/fluid_build/build_runners/_deployment_host_policy.py index 604157bd9..3dc46ae6e 100644 --- a/fluid_build/build_runners/_deployment_host_policy.py +++ b/fluid_build/build_runners/_deployment_host_policy.py @@ -77,7 +77,7 @@ import os from dataclasses import dataclass -from typing import Any, Callable, List, Mapping, Optional +from typing import Any, Callable, List, Literal, Mapping, Optional from urllib.parse import urlsplit from fluid_build._errors import FluidUserError, doc_url @@ -127,6 +127,105 @@ def _on_allowlist(host: str, suffixes: List[str]) -> bool: return False +VerdictCode = Literal["missing", "scheme", "no-host", "link-local", "mixed", "public"] + + +@dataclass(frozen=True) +class HostVerdict: + """Why the policy refuses a URL, without wording. + + Each caller phrases ``code`` for its own case, so the decision has one home + and only the message differs. ``pinned`` is the address a request was + pinned to, when that address and not the name decided. + """ + + code: VerdictCode + host: Optional[str] = None + scheme: str = "" + pinned: Optional[str] = None + + +def host_verdict( + url: Optional[str], + *, + environ: Optional[Mapping[str, str]] = None, + operator_supplied: bool = False, +) -> Optional[HostVerdict]: + """The policy's decision for ``url``: ``None`` when it may be used. + + :func:`host_refusal` words it for an acquisition orchestrator; the managed + Flink lane words it for an Iceberg catalog (``streaming/policy.py``). + """ + if not url or not str(url).strip(): + return HostVerdict("missing") + try: + parts = urlsplit(str(url).strip()) + host = parts.hostname + except ValueError: # an unbalanced "[" in the netloc: no host can be read from it + return HostVerdict("no-host") + scheme = (parts.scheme or "").lower() + if scheme not in _ALLOWED_SCHEMES: + return HostVerdict("scheme", scheme=scheme) + if not host: + return HostVerdict("no-host", scheme=scheme) + # First, and not overridable: this is where instance metadata lives, and no + # orchestrator is ever legitimately there. + if _hostname_is_link_local(host): + return HostVerdict("link-local", host, scheme) + 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 HostVerdict("mixed", host, scheme) + 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 HostVerdict("public", host, scheme) + + +def pinned_verdict( + url: Optional[str], + pinned: Optional[str], + *, + environ: Optional[Mapping[str, str]] = None, + operator_supplied: bool = False, +) -> Optional[HostVerdict]: + """The decision for the address a request is actually pinned to. + + A request hook added to ``safe_httpx_client`` runs after its pinning hook, + which has replaced the hostname with the IP the connection will use. + Judging that IP closes two gaps that resolving the name again leaves open: + DNS rebinding between the two lookups, and a parser differential, where + ``socket.getaddrinfo`` encodes ``faß.de`` as IDNA2003 ``fass.de`` while + httpx resolves IDNA2008 ``xn--fa-hia.de``. The name ``url`` carries still + decides the allowlist; the pinned address decides the scope. A refusal + names both, because the name is what an allowlist entry would list. + """ + if not pinned: + return HostVerdict("no-host") + try: + host = urlsplit(str(url or "").strip()).hostname + except ValueError: + host = None + if _hostname_is_link_local(pinned): + return HostVerdict("link-local", host or pinned, pinned=pinned) + if operator_supplied or (host and _on_allowlist(host, allowlist(environ))): + return None + if hostname_scope(pinned) == "non_public": + return None + return HostVerdict("public", host or pinned, pinned=pinned) + + def host_refusal( url: Optional[str], *, @@ -149,52 +248,32 @@ def host_refusal( ``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(): + verdict = host_verdict(url, environ=environ, operator_supplied=operator_supplied) + if verdict is None: + return None + if verdict.code == "missing": return "no deployment.server_url is configured" - parts = urlsplit(str(url).strip()) - scheme = (parts.scheme or "").lower() - if scheme not in _ALLOWED_SCHEMES: + if verdict.code == "scheme": return ( - f"its scheme is {scheme or 'missing'}; the {engine} orchestrator must be an " + f"its scheme is {verdict.scheme or 'missing'}; the {engine} orchestrator must be an " "http:// or https:// URL" ) - try: - host = parts.hostname - except ValueError: - host = None - if not host: + if verdict.code == "no-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): + host = verdict.host + if verdict.code == "link-local": 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": + if verdict.code == "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" @@ -218,8 +297,13 @@ def refuse_unless_host_allowed( refusal = host_refusal(url, engine=engine, environ=environ, operator_supplied=operator_supplied) if refusal is None: return str(url).strip() + raise _host_error(refusal, engine=engine, operator_supplied=operator_supplied) + + +def _host_error(refusal: str, *, engine: str, operator_supplied: bool) -> DeploymentHostError: + """The :class:`DeploymentHostError` for ``refusal``, worded for whoever chose the URL.""" if operator_supplied: - raise DeploymentHostError( + return DeploymentHostError( what=f"the {engine} endpoint given on the command line cannot be used", why=f"Refused because {refusal}.", fix=( @@ -229,7 +313,7 @@ def refuse_unless_host_allowed( ), doc=doc_url(), ) - raise DeploymentHostError( + return 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 " @@ -248,24 +332,58 @@ def refuse_unless_host_allowed( ) +def _pinned_refusal(verdict: HostVerdict) -> str: + """:func:`host_refusal`'s words for a :func:`pinned_verdict`, naming both. + + The host is what an allowlist entry would list. The address is where the + connection was going, and the only part that tells a rebinding answer from + a host that really is public. + """ + if verdict.code == "no-host": + return "the request has no address to check" + named = f"host {verdict.host!r}" + if verdict.pinned and verdict.pinned != verdict.host: + named += f", connecting to {verdict.pinned!r}," + if verdict.code == "link-local": + return ( + f"{named} is link-local or unspecified, where cloud instance metadata lives; " + "no orchestrator is there, and this is refused even with an allowlist entry" + ) + return ( + f"{named} is public, and the config posted to it carries the source credential " + "in clear text" + ) + + 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. + """An httpx request hook that judges the address the request is pinned to. 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. + ``safe_httpx_client`` inserts its DNS-pinning hook ahead of this one, and + that hook rewrites ``request.url`` to the IP the connection will use. This + judges that IP rather than resolving the name again: a second lookup can + disagree with the first (DNS rebinding) or encode a non-ASCII name + differently (Python's IDNA2003 against httpx's IDNA2008). The name in + ``url``, closed over at construction, is still checked as it was there and + still decides the allowlist, because the hostname is what an entry names. """ - def _guard(request: Any) -> None: # pragma: no cover - exercised via the clients + def _guard(request: Any) -> None: refuse_unless_host_allowed(url, engine=engine, operator_supplied=operator_supplied) + verdict = pinned_verdict( + url, + getattr(getattr(request, "url", None), "host", None), + operator_supplied=operator_supplied, + ) + if verdict is not None: + raise _host_error( + _pinned_refusal(verdict), engine=engine, operator_supplied=operator_supplied + ) return _guard diff --git a/pyproject.toml b/pyproject.toml index 250800598..8ea9492b1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -808,6 +808,12 @@ module = [ # becomes in the config forge-cli pushes, so a silent Any is a silent leak. "fluid_build.providers._iceberg_catalog_auth", "fluid_build.secrets", + # Operator egress gates: which host a contract-named URL may reach, which + # variable may be sent there, and where a managed Flink job may deploy. A + # silent Any is a silent allow, and strict equality is what makes a + # misspelt verdict code fail here instead of falling through. + "fluid_build.build_runners._deployment_host_policy", + "fluid_build.build_runners._secret_policy", ] strict = true diff --git a/tests/_infrastructure/respx_fixtures.py b/tests/_infrastructure/respx_fixtures.py index 58b4e5cf4..74e04d19f 100644 --- a/tests/_infrastructure/respx_fixtures.py +++ b/tests/_infrastructure/respx_fixtures.py @@ -46,11 +46,26 @@ def requires_respx(reason: str = "respx not installed") -> Any: return pytest.mark.skipif(respx is None, reason=reason) +def _list_orchestrator(monkeypatch: pytest.MonkeyPatch, host: str) -> None: + """Name a mock orchestrator in the operator allowlist, as its owner would. + + The unit suite replaces safe_http's pin hook with a no-op (tests/conftest.py), + so the deployment-host request guard sees this name where production sees + the IP the request was pinned to. A name that does not resolve is no address + the guard can allow, so it is refused unless listed. Listing it keeps the + link-local refusal in force. + """ + from fluid_build.build_runners._deployment_host_policy import ALLOWLIST_ENV + + monkeypatch.setenv(ALLOWLIST_ENV, host) + + @pytest.fixture -def airbyte_mock() -> Iterator["AirbyteMockServer"]: +def airbyte_mock(monkeypatch: pytest.MonkeyPatch) -> Iterator["AirbyteMockServer"]: """In-memory Airbyte API. Tests mutate state via ``airbyte_mock.create_source(...)``; runner code under test sees realistic CRUD responses. """ + _list_orchestrator(monkeypatch, "airbyte.test") server = AirbyteMockServer() with respx.mock(base_url="https://airbyte.test", assert_all_called=False) as router: server.attach(router) @@ -58,7 +73,8 @@ def airbyte_mock() -> Iterator["AirbyteMockServer"]: @pytest.fixture -def kafka_connect_mock() -> Iterator["KafkaConnectMockServer"]: +def kafka_connect_mock(monkeypatch: pytest.MonkeyPatch) -> Iterator["KafkaConnectMockServer"]: + _list_orchestrator(monkeypatch, "kafka-connect.test") server = KafkaConnectMockServer() with respx.mock(base_url="http://kafka-connect.test:8083", assert_all_called=False) as router: server.attach(router) diff --git a/tests/build_runners/test_deployment_host_policy.py b/tests/build_runners/test_deployment_host_policy.py index bc9bfe20a..a69868503 100644 --- a/tests/build_runners/test_deployment_host_policy.py +++ b/tests/build_runners/test_deployment_host_policy.py @@ -47,6 +47,10 @@ 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. +4. **The guard judges the address the request is pinned to.** It resolved the + name a second time, so DNS rebinding, or ``faß.de`` looked up as IDNA2003 + ``fass.de`` while httpx connects to IDNA2008 ``xn--fa-hia.de``, could pass + a name whose connection went to an unlisted public address or to metadata. """ from __future__ import annotations @@ -56,6 +60,7 @@ import socket import threading from http.server import BaseHTTPRequestHandler, HTTPServer +from types import SimpleNamespace from typing import Any, Dict, Iterator, List, Tuple import pytest @@ -81,6 +86,11 @@ "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"], + # Private when asked a second time, whatever the pinning lookup got. + "orchestrator.rebind.test": ["10.1.2.3"], + # What socket.getaddrinfo answers for faß.de: it asks about IDNA2003 + # fass.de, while httpx connects to IDNA2008 xn--fa-hia.de. + "faß.de": ["10.1.2.3"], } @@ -227,6 +237,7 @@ def test_the_address_predicate_is_shared_not_copied() -> None: ("file:///etc/passwd", "", False, "not an http(s) orchestrator"), ("", "", False, "nothing configured"), ("https://", "", False, "no host"), + ("http://[::1", "", False, "an unbalanced bracket has no host"), ], ) def test_the_policy_matrix(url: str, listed: str, allowed: bool, because: str) -> None: @@ -397,21 +408,77 @@ def log_message(self, *_a: Any) -> None: 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.""" +def _pinned(address: str) -> Any: + """A request as the guard sees it: safe_httpx_client's pin hook has put the IP in.""" + return SimpleNamespace(url=SimpleNamespace(host=address)) - class _Explodes: - @property - def url(self) -> str: - raise AssertionError("the guard must not read the rewritten request URL") - import os +@pytest.mark.parametrize( + "url, pinned, listed, refused", + [ + # The guard's own lookup of the name says private (rebinding, or the + # IDNA2003 spelling of faß.de), yet the connection is pinned to a + # public address: refused. + ("https://orchestrator.rebind.test", "93.184.216.34", "", True), + ("https://faß.de", "93.184.216.34", "", True), + ("http://orchestrator.rebind.test:8083", "10.1.2.3", "", False), + # A listed name may connect to its public address, never to metadata. + ("https://orchestrator.public.test", "93.184.216.34", "orchestrator.public.test", False), + ("https://orchestrator.public.test", "169.254.169.254", "orchestrator.public.test", True), + ("https://orchestrator.public.test", "0.0.0.0", "orchestrator.public.test", True), + # No address to judge is not a pass. + ("https://orchestrator.public.test", "", "orchestrator.public.test", True), + ], +) +def test_the_guard_judges_the_pinned_address_not_a_second_lookup( + monkeypatch: pytest.MonkeyPatch, url: str, pinned: str, listed: str, refused: bool +) -> None: + """The pin hook runs first and puts the IP the connection will use in the URL. - os.environ[ALLOWLIST_ENV] = "orchestrator.public.test" - try: - request_guard("https://orchestrator.public.test", engine="airbyte")(_Explodes()) - finally: - del os.environ[ALLOWLIST_ENV] + Resolving the name again can answer differently: DNS rebinding between the + two lookups, or Python's IDNA2003 against httpx's IDNA2008. The name still + decides the allowlist; the pinned address decides the scope. + """ + monkeypatch.setenv(ALLOWLIST_ENV, listed) + guard = request_guard(url, engine="kafka-connect") + if refused: + with pytest.raises(DeploymentHostError): + guard(_pinned(pinned)) + else: + guard(_pinned(pinned)) + + +def test_a_pinned_refusal_names_the_host_to_list_and_the_address_it_reached() -> None: + with pytest.raises(DeploymentHostError) as excinfo: + request_guard("https://orchestrator.rebind.test", engine="debezium")( + _pinned("93.184.216.34") + ) + error = excinfo.value + assert "'orchestrator.rebind.test', connecting to '93.184.216.34'," in (error.why or "") + assert "debezium" in (error.why or "") + assert f'{ALLOWLIST_ENV}="connect.example.com"' in (error.fix or "") + + +def test_an_operator_chosen_url_still_never_connects_to_metadata() -> None: + guard = request_guard( + "https://orchestrator.public.test", engine="airbyte", operator_supplied=True + ) + guard(_pinned("93.184.216.34")) + with pytest.raises(DeploymentHostError) as excinfo: + guard(_pinned("169.254.169.254")) + error = excinfo.value + rendered = " ".join([str(error), error.why or "", error.fix or ""]) + assert "connecting to '169.254.169.254'" in rendered + assert "contract" not in rendered, rendered + + +def test_a_malformed_url_is_refused_not_raised() -> None: + """``http://[::1`` made ``urlsplit`` raise ValueError through the policy.""" + with pytest.raises(DeploymentHostError) as excinfo: + refuse_unless_host_allowed("http://[::1", engine="kafka-connect") + assert "no host" in (excinfo.value.why or "") + with pytest.raises(DeploymentHostError): + request_guard("http://[::1", engine="kafka-connect")(_pinned("127.0.0.1")) @pytest.mark.parametrize( diff --git a/tests/build_runners/test_deployment_host_policy_real_pin_hook.py b/tests/build_runners/test_deployment_host_policy_real_pin_hook.py new file mode 100644 index 000000000..08b48a81b --- /dev/null +++ b/tests/build_runners/test_deployment_host_policy_real_pin_hook.py @@ -0,0 +1,123 @@ +# 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. + +"""The deployment-host guard judges what the REAL pin hook pinned. + +``tests/conftest.py`` replaces safe_http's pin hook with a no-op for the unit +suite, so ``test_deployment_host_policy.py`` hands the guard a stand-in +request. That cannot catch the one ordering the fix relies on: the pin hook +runs first and writes the IP the connection will use into ``request.url``, +and the guard reads it there. This file is on that fixture's real-guard list, +so the clients below run the real ``safe_httpx_client`` stack. A guard moved +ahead of the pin hook, or one that resolves the name again, lets the request +reach the transport and fails here. + +DNS is faked and the transport records instead of sending, so nothing leaves +the machine. +""" + +from __future__ import annotations + +import socket +import sys +from typing import Any, Iterator, List, Tuple + +import httpx +import pytest + +from fluid_build.build_runners._deployment_host_policy import ALLOWLIST_ENV, DeploymentHostError + +PUBLIC = "93.184.216.34" +PRIVATE = "10.1.2.3" + +_REAL_GETADDRINFO = socket.getaddrinfo + + +def _answer(address: str) -> List[Tuple[Any, ...]]: + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, 0))] + + +def _getaddrinfo(host: str, *a: Any, **k: Any) -> List[Tuple[Any, ...]]: + if host == "rebind.test": + # The pin hook's lookup gets the attacker's public answer; every + # lookup the policy makes of its own gets a private one. + pinning = sys._getframe(1).f_code.co_name == "resolve_and_validate" + return _answer(PUBLIC if pinning else PRIVATE) + if host == "faß.de": # what Python's resolver asks about: IDNA2003 fass.de + return _answer(PRIVATE) + if host == "xn--fa-hia.de": # what httpx connects to: IDNA2008 + return _answer(PUBLIC) + if host == "connect.internal": + return _answer(PRIVATE) + return _REAL_GETADDRINFO(host, *a, **k) + + +@pytest.fixture(autouse=True) +def sent(monkeypatch: pytest.MonkeyPatch) -> Iterator[List[str]]: + """The URLs that reached the transport, which answers instead of sending.""" + reached: List[str] = [] + + def _record(self: Any, request: httpx.Request) -> httpx.Response: + reached.append(str(request.url)) + return httpx.Response(200, json={}, request=request) + + monkeypatch.setattr(socket, "getaddrinfo", _getaddrinfo) + monkeypatch.setattr(httpx.HTTPTransport, "handle_request", _record) + monkeypatch.delenv(ALLOWLIST_ENV, raising=False) + yield reached + + +@pytest.mark.parametrize( + "url", + [ + "http://rebind.test:8083", # DNS rebinding between the two lookups + "http://faß.de:8083", # IDNA2003 against httpx's IDNA2008 + ], +) +def test_a_name_that_looks_private_twice_is_refused_at_its_pinned_address( + url: str, sent: List[str] +) -> None: + from fluid_build.build_runners.kafka_connect.runner import KafkaConnectRestClient + + client = KafkaConnectRestClient(url) # every check of the name says private + try: + with pytest.raises(DeploymentHostError) as excinfo: + client.list_connectors() + finally: + client.close() + assert sent == [], f"the config must not leave for {PUBLIC}, got {sent}" + assert f"connecting to '{PUBLIC}'" in (excinfo.value.why or "") + + +def test_the_airbyte_client_is_guarded_the_same_way(sent: List[str]) -> None: + from fluid_build.build_runners.airbyte.runner import AirbyteRestClient + + client = AirbyteRestClient("https://rebind.test", api_token="t") + try: + with pytest.raises(DeploymentHostError): + client.list_sources("workspace") + finally: + client.close() + assert sent == [] + + +def test_a_private_orchestrator_is_reached_at_its_pinned_address(sent: List[str]) -> None: + from fluid_build.build_runners.kafka_connect.runner import KafkaConnectRestClient + + client = KafkaConnectRestClient("http://connect.internal:8083") + try: + client.list_connectors() + finally: + client.close() + assert sent == [f"http://{PRIVATE}:8083/connectors"] diff --git a/tests/conftest.py b/tests/conftest.py index 0eaf60cbb..0657541d9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -359,6 +359,7 @@ def _disable_ssrf_guard_for_unit_tests(request): _real_guard_files = ( "tests/util/test_safe_http.py", "tests/test_forge_web_tools.py", + "tests/build_runners/test_deployment_host_policy_real_pin_hook.py", ) if any(marker in fspath for marker in _real_guard_files): yield From 55e64c900241873d5066122738ea82c66bd3b19d Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 4 Nov 2026 05:36:34 +0100 Subject: [PATCH 2/6] security: the ODCS and plugin-extension checks let a trailing newline through a pattern (#811) * security: the ODCS and plugin-extension checks let a trailing newline through a pattern The FLUID fix (schema_validator, #799) left six JSON Schema checks on jsonschema's own pattern keyword, Python's re.search, whose $ also matches before a final newline: - forge/core/artifact_validators.py (fluid validate-artifacts: ODCS, Bitol ODPS) - providers/odcs/validation.py::collect_errors (fluid odcs validate) - providers/odps_standard/validation.py::validate - providers/opds/validator.py (pinned Draft202012Validator) - forge_datamodel/emit/osi_sidecar.py (the copilot's Ossie check) - extension_schemas.py::assemble_proposed_extensions (pinned Draft7Validator) So ODCS refused schema[].id "t 1" but took "t1\n", which an id: | block produces, and the copilot kept an LLM-proposed customScaffold block with use: "ci:basic\n", ref: "main\n" or id: "ci\n". All six now go through schema_validator, and say so with the same line-break hint. schema_validator takes a default dialect for a schema that declares none, passed to jsonschema's validator_for(default=), the way OpenStack Rally uses it. An extension schema is read in the dialect it declares, draft-07 when it declares none (the documented plugin contract), so a 2020-12 schema's unevaluatedProperties: false is enforced instead of ignored. OPDS keeps 2020-12 as its default. patternProperties keys keep re.search. jsonschema matches them in patternProperties, additionalProperties and unevaluatedProperties, and check-jsonschema's override of patternProperties alone lets a key that one keyword matches and another does not through additionalProperties: false unchecked. Only OPDS has key patterns (^[a-z]{2}$, three places). The tripwires now cover every shipped schema, pin that set, and also fail on a $recursiveRef or $dynamicRef, which re-pick the stock class as a nested $schema does. Follows #799. * security: review follow-ups for the vendored-schema end-anchor change - fluid odcs validate cuts each schema message to 120 characters, so for an id like customer_orders_v1 the hint ended "... block keeps and |", which reads as advice to use |. _brief cuts the message ahead of the hint instead; the hint is schema_manager.LINE_BREAK_HINT now, so the command can find it. - schema_validator raises ImportError when jsonschema could not be imported. validate_osi_document, the one unguarded caller, raised NameError: jsonschema instead. validate_opds reads the same JSONSCHEMA_AVAILABLE flag as the other checks; the find_spec probe that found an unimportable jsonschema is gone. - A test walks fluid_build/ and fails on any validator_for(), Draft*Validator() or jsonschema.validate() without cls= outside schema_manager, so a check added later cannot read $ as Python does again. It finds the six sites on the base and seven on the old main. - Tests for the flag branches of collect_errors and the artifact validators, and for OPDS reading an undeclared schema as 2020-12. - CHANGELOG: the extension check shows no message, so it carries no hint. - odps_standard.validate validates against the schema for the document's apiVersion, not only v1.0.0. * fix: a bracket in a value or an unusable plugin schema no longer stops the run Defect-scan follow-ups for the vendored-schema end-anchor change: - fluid odcs validate printed each schema error as Rich markup, so a failing value with "[/]" in it ended the command with "Unexpected error: closing tag '[/]' ... has nothing to close" (exit 2, also on main). The line is printed with markup=False, like the vowl line in the same function. - assemble_proposed_extensions raised when a plugin schema's $schema was not a string: validator_for cannot look it up, where Draft7Validator ignored it. A schema that cannot be applied now drops its block with a redacted warning, as a block that fails its schema is dropped, and the run goes on. This also covers {"type": 5}, which raised on main. - The guard test also catches `from jsonschema import validate`, `import jsonschema as js` with js.validate(...), jsonschema.validators.validate/extend/create, and a cls= that does not come from schema_validator. ODPS passes cls=type(schema_validator(...)) inline so the guard can see it. It parses only files that mention jsonschema. * docs(changelog): keep the two pattern end-anchor entries together The rebase onto main put two new fluid diff --baseline entries between the FLUID-schema entry (#799) and this branch's, which begins "The fix above covered the FLUID schemas only". --- CHANGELOG.md | 17 + fluid_build/cli/odcs.py | 12 +- fluid_build/extension_schemas.py | 18 +- fluid_build/forge/core/artifact_validators.py | 21 +- .../forge_datamodel/emit/osi_sidecar.py | 8 +- fluid_build/providers/odcs/validation.py | 12 +- .../providers/odps_standard/validation.py | 9 +- fluid_build/providers/opds/validator.py | 8 +- fluid_build/schema_manager.py | 32 +- tests/test_schema_pattern_end_anchor.py | 48 ++- ...test_standard_schema_pattern_end_anchor.py | 315 ++++++++++++++++++ 11 files changed, 454 insertions(+), 46 deletions(-) create mode 100644 tests/test_standard_schema_pattern_end_anchor.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ec3befd4c..8c27efb8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -275,6 +275,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 as before. The Command Center run now reports `apply_build_id_not_found`, where it reported `build_not_found:`. +- **`fluid odcs validate` stopped with `Unexpected error: closing tag '[/]' ... has nothing to + close`** when a value that fails a schema check contained `[/]` or another Rich markup tag. The + error line quotes the value, and it was printed as markup. It is printed as plain text now, so + the command reports the error and exits 1. + - **A missing optional dependency names the extra to install, and silent fallbacks warn.** litellm, the MCP SDK, openlineage, boto3 and duckdb are imported through one gate: when one is missing, the command prints ` needs , which is not installed.` and @@ -620,6 +625,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 and |- or >- drops)`. A `sink.catalogAuth` field gets no hint, since its value is never shown. The schemas themselves are unchanged. +- **The ODCS and plugin-extension checks let a value with a trailing newline through a `pattern`, + too.** The fix above covered the FLUID schemas only. `fluid odcs validate` and + `fluid validate-artifacts` accepted an ODCS `schema[].id: "t1\n"` while refusing `"t 1"`, and + the forge copilot kept an LLM-proposed extension block whose value ended in a line break, such + as `use: "ci:basic\n"` for `customScaffold`. Every JSON Schema check now reads `$` as the end of + the value, including the ODPS, OPDS and Ossie checks, whose vendored schemas have no such + pattern yet. Their errors carry the same hint; an extension block that fails is dropped, as + before. An extension schema is read in the dialect its `$schema` declares, and as draft-07 when + it declares none, so a 2020-12 `unevaluatedProperties: false` is no longer ignored. + `patternProperties` keys are still matched as before, so OPDS's language-code key `^[a-z]{2}$` + accepts `"en\n"`. + - **`fluid diff --baseline` compared contracts that no schema accepts.** Its rules read only the shapes a schema allows, so a value of the wrong type hid a change: a column retyped from `decimal(18,2)` to the number `18` passed `--fail-on-breaking` with exit 0. Both contracts are diff --git a/fluid_build/cli/odcs.py b/fluid_build/cli/odcs.py index b277d913e..bebc294e3 100644 --- a/fluid_build/cli/odcs.py +++ b/fluid_build/cli/odcs.py @@ -504,6 +504,15 @@ def _validate_imported_contract(contract: dict) -> list: return [] if result.is_valid else list(result.errors) +def _brief(message: str, limit: int = 120) -> str: + """``message`` cut to ``limit`` characters, keeping a line-break hint whole: cut, + ``... block keeps and |`` reads as advice to use ``|``.""" + from fluid_build.schema_manager import LINE_BREAK_HINT + + head = message.removesuffix(LINE_BREAK_HINT) + return head[:limit] + message[len(head) :] + + def _run_odcs_validate(args): """Run ODCS validate command — multi-pass with structured report. @@ -559,7 +568,8 @@ def _run_odcs_validate(args): report["ok"] = False cprint(f"✗ jsonschema: {len(errors)} error(s)") for e in errors[:10]: - cprint(f" {e['path'] or ''}: {e['message'][:120]}") + # markup=False: the message quotes the value, and "[/]" in it is a Rich tag. + cprint(f" {e['path'] or ''}: {_brief(e['message'])}", markup=False) if len(errors) > 10: cprint(f" ... +{len(errors) - 10} more") else: diff --git a/fluid_build/extension_schemas.py b/fluid_build/extension_schemas.py index db8304993..98a3555ce 100644 --- a/fluid_build/extension_schemas.py +++ b/fluid_build/extension_schemas.py @@ -22,7 +22,8 @@ customScaffold = "data_product_forge_custom_scaffold.validation:get_extension_schema" The provider has the signature ``get_extension_schema(fluid_version=None) -> dict`` -and returns a draft-07 JSON Schema describing the data *under* the extension key. +and returns a JSON Schema describing the data *under* the extension key: draft-07, +unless the schema declares another dialect in its ``$schema``. These helpers let the ``fluid forge`` copilot (a) ground contract generation on the installed extension schemas and (b) validate the generated blocks before @@ -200,6 +201,7 @@ def assemble_proposed_extensions( from jsonschema import Draft7Validator except ImportError: # pragma: no cover — jsonschema is a hard dependency Draft7Validator = None # type: ignore[assignment] + from fluid_build.schema_manager import schema_validator out: Dict[str, Any] = {} for key, block in proposed.items(): @@ -207,7 +209,19 @@ def assemble_proposed_extensions( if schema is None: continue if Draft7Validator is not None: - if next(iter(Draft7Validator(schema).iter_errors(block)), None) is not None: + # The dialect the plugin's schema declares, draft-07 when it declares + # none; Draft 7 ignores a newer keyword rather than enforcing it. A + # pattern's "$" is the end of the value, as ECMA-262 reads it. + try: + validator = schema_validator(schema, default=Draft7Validator) + invalid = next(iter(validator.iter_errors(block)), None) is not None + except Exception as e: # a schema that cannot be applied, e.g. a non-string $schema + from fluid_build.observability.secret_redactor import redact_secret_text + + log = logger or logging.getLogger(__name__) + log.warning("%s", redact_secret_text(f"extension schema {key!r} unusable: {e}")) + continue # unchecked — don't emit it + if invalid: continue # invalid — let the repair loop surface it; don't emit junk out[key] = block return out diff --git a/fluid_build/forge/core/artifact_validators.py b/fluid_build/forge/core/artifact_validators.py index 5577ffd9c..1d0c99f97 100644 --- a/fluid_build/forge/core/artifact_validators.py +++ b/fluid_build/forge/core/artifact_validators.py @@ -36,7 +36,6 @@ from __future__ import annotations import hashlib -import importlib.util import json import logging import shutil @@ -113,10 +112,6 @@ def _load_schema(path: Path) -> Optional[Dict[str, Any]]: # --------------------------------------------------------------------------- -def _jsonschema_available() -> bool: - return importlib.util.find_spec("jsonschema") is not None - - def _conftest_available() -> bool: """``conftest`` binary on PATH — independent of python package install.""" return shutil.which("conftest") is not None @@ -306,7 +301,10 @@ def _validate_against_schema( ) ] - if not _jsonschema_available(): + from fluid_build.schema_manager import JSONSCHEMA_AVAILABLE, schema_validator + + # The flag schema_validator needs: a package that is found can still fail to import. + if not JSONSCHEMA_AVAILABLE: return [ ValidationIssue( file=artifact_path, @@ -320,8 +318,6 @@ def _validate_against_schema( ) ] - import jsonschema # type: ignore - issues: List[ValidationIssue] = [] # Validate with the dialect the schema DECLARES, not a hardcoded one. # @@ -335,7 +331,10 @@ def _validate_against_schema( # and guards nine objects with `unevaluatedProperties: false`. A server entry # carrying an unexpected key validated with ZERO errors, reachable from # `fluid validate-artifacts`. See tests/test_jsonschema_dialect.py. - validator = jsonschema.validators.validator_for(schema)(schema) + # + # schema_validator also reads a pattern's `$` as the end of the value, as + # ECMA-262 does: ODCS refused `schema[].id: "t 1"` but took `"t1\n"`. + validator = schema_validator(schema) errors = sorted(validator.iter_errors(doc), key=lambda e: list(e.absolute_path)) for err in errors: @@ -397,7 +396,9 @@ def validate_opds(path: str, content: bytes) -> List[ValidationIssue]: ) ] - if not _jsonschema_available(): + from fluid_build.schema_manager import JSONSCHEMA_AVAILABLE + + if not JSONSCHEMA_AVAILABLE: return [ ValidationIssue( file=path, diff --git a/fluid_build/forge_datamodel/emit/osi_sidecar.py b/fluid_build/forge_datamodel/emit/osi_sidecar.py index d9d5f2f10..206749753 100644 --- a/fluid_build/forge_datamodel/emit/osi_sidecar.py +++ b/fluid_build/forge_datamodel/emit/osi_sidecar.py @@ -93,14 +93,14 @@ def validate_osi_document(document: Dict[str, Any]) -> List[str]: Returns a sorted list of human-readable issue strings (empty when conformant). ``jsonschema`` is imported lazily — this module sits on the forge pipeline path, not the CLI cold path, but there is no - reason to tax importers that never validate. + reason to tax importers that never validate. A ``pattern``'s ``$`` is + the end of the value, as ECMA-262 reads it. """ - import jsonschema + from fluid_build.schema_manager import schema_validator schema = _load_ossie_schema() - validator_cls = jsonschema.validators.validator_for(schema) errors = sorted( - validator_cls(schema).iter_errors(document), + schema_validator(schema).iter_errors(document), key=lambda e: ([str(p) for p in e.absolute_path], e.message), ) return [f"{'/'.join(str(p) for p in e.absolute_path) or ''}: {e.message}" for e in errors] diff --git a/fluid_build/providers/odcs/validation.py b/fluid_build/providers/odcs/validation.py index 7cb2260db..9711f5822 100644 --- a/fluid_build/providers/odcs/validation.py +++ b/fluid_build/providers/odcs/validation.py @@ -74,15 +74,17 @@ def collect_errors(odcs: Mapping[str, Any], schema: Mapping[str, Any]) -> List[D Empty list = valid. No-op (empty list) when :mod:`jsonschema` is not installed. Used by ``validate()`` (which raises on non-empty) and by ``fluid odcs validate --report`` (which serializes to JSON for CI). + + A ``pattern``'s ``$`` is the end of the value, as ECMA-262 reads it, so + ``schema[].id: "t1\\n"`` fails ``^[A-Za-z0-9_-]+$`` like ``"t 1"`` does. """ - try: - import jsonschema - except ImportError: + from fluid_build.schema_manager import JSONSCHEMA_AVAILABLE, schema_validator + + if not JSONSCHEMA_AVAILABLE: LOG.warning("jsonschema not installed, skipping ODCS validation") return [] - validator_cls = jsonschema.validators.validator_for(schema) - validator = validator_cls(schema) + validator = schema_validator(schema) out: List[Dict[str, Any]] = [] for err in validator.iter_errors(odcs): out.append( diff --git a/fluid_build/providers/odps_standard/validation.py b/fluid_build/providers/odps_standard/validation.py index b6428c2ea..7f33d6bca 100644 --- a/fluid_build/providers/odps_standard/validation.py +++ b/fluid_build/providers/odps_standard/validation.py @@ -75,13 +75,18 @@ def schema_for_document(odps: Mapping[str, Any]) -> Optional[Dict[str, Any]]: def validate(odps: Mapping[str, Any], schema: Mapping[str, Any]) -> None: - """Validate an ODPS product against the vendored v1.0.0 JSON Schema.""" + """Validate an ODPS product against ``schema``, the vendored one for its + ``apiVersion`` (:func:`schema_for_document`).""" try: import jsonschema except ImportError: LOG.warning("jsonschema not installed, skipping ODPS validation") return + from fluid_build.schema_manager import schema_validator + try: - jsonschema.validate(instance=odps, schema=schema) + # cls: the declared dialect, with a pattern's "$" read as the end of the + # value, as ECMA-262 does. jsonschema.validate still checks the schema first. + jsonschema.validate(instance=odps, schema=schema, cls=type(schema_validator(schema))) except jsonschema.ValidationError as exc: raise ProviderError(f"ODPS validation failed: {exc.message}") from exc diff --git a/fluid_build/providers/opds/validator.py b/fluid_build/providers/opds/validator.py index 1f131cf07..c75b00803 100644 --- a/fluid_build/providers/opds/validator.py +++ b/fluid_build/providers/opds/validator.py @@ -119,8 +119,14 @@ def validate_against_opds_schema( valid, errors = _basic_validation(opds_data, version) return valid, errors, "basic" + from fluid_build.schema_manager import schema_validator + try: - validator = jsonschema.Draft202012Validator(schema) + # The dialect the schema declares (v4.1: 2020-12, also the default), + # with a pattern's "$" read as the end of the value, as ECMA-262 does. + # The patternProperties language-code keys (^[a-z]{2}$) keep + # jsonschema's own re.search; see schema_manager.schema_validator. + validator = schema_validator(schema, default=jsonschema.Draft202012Validator) errors = list(validator.iter_errors(opds_data)) # Filter a known upstream OPDS v4.1 schema false-positive: the diff --git a/fluid_build/schema_manager.py b/fluid_build/schema_manager.py index 1b992d7ee..e4f0b0c7b 100644 --- a/fluid_build/schema_manager.py +++ b/fluid_build/schema_manager.py @@ -28,7 +28,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta from pathlib import Path -from typing import Any, Callable, Dict, FrozenSet, Iterator, List, Optional, Union +from typing import Any, Callable, Dict, FrozenSet, Iterator, List, Mapping, Optional, Union import yaml @@ -43,6 +43,7 @@ # anywhere in this module — and RefResolver is deprecated, so the day a # jsonschema release drops it this ImportError would have silently disabled # schema validation across the CLI. Never name more in a probe than you use. +_JSONSCHEMA_IMPORT_ERROR: Optional[ImportError] = None try: import jsonschema @@ -66,6 +67,9 @@ #: Validators whose jsonschema message names only keys, never a value. _KEY_ONLY_VALIDATORS = frozenset({"required", "additionalProperties"}) +#: Ends a ``pattern`` error when the value's final line break is all that fails it. +LINE_BREAK_HINT = " (it ends in a line break, which a YAML | or > block keeps and |- or >- drops)" + def format_schema_error(error: Any) -> str: """``error.message``, without the failing value under :data:`_UNQUOTED_VALUE_PATHS`. @@ -175,7 +179,7 @@ def _pattern_keyword(validator: Any, pattern: str, instance: Any, schema: Any) - return message = f"{instance!r} does not match {pattern!r}" if re.search(pattern, instance): # only the final line break fails - message += " (it ends in a line break, which a YAML | or > block keeps and |- or >- drops)" + message += LINE_BREAK_HINT yield jsonschema.ValidationError(message) @@ -184,20 +188,32 @@ def _end_anchored_validator(validator_class: Any) -> Any: return jsonschema.validators.extend(validator_class, {"pattern": _pattern_keyword}) -def schema_validator(schema: Dict[str, Any]) -> Any: +def schema_validator(schema: Mapping[str, Any], default: Any = None) -> Any: """A jsonschema validator for ``schema``, in the dialect it declares, whose ``pattern`` ends at the end of the value (:func:`_end_anchored`). Validate - against a FLUID schema through this, never ``validator_for`` directly. + against a FLUID schema, or a vendored or plugin one, through this, never + ``validator_for`` directly. ``default`` is the dialect of a schema that + declares none, jsonschema's newest when not given. Raises ``ImportError`` + when jsonschema could not be imported; check :data:`JSONSCHEMA_AVAILABLE` + first to skip instead. The keyword is replaced the way check-jsonschema replaces it, with ``jsonschema.validators.extend``. Two places keep jsonschema's own ``re.search``: ``patternProperties`` keys, as its ``additionalProperties`` and :func:`_locate_error` read them, and a subschema that declares its own - ``$schema`` (or a ``$ref`` to the root, which does), for which jsonschema - picks that dialect's own class. No FLUID schema has either; a test fails if - one does. + ``$schema`` (or a ``$ref``, ``$recursiveRef`` or ``$dynamicRef`` to the + root, which does), for which jsonschema picks that dialect's own class. No + FLUID schema has either. Of the vendored ones, only OPDS keys + ``patternProperties``, with ``^[a-z]{2}$`` language codes. A test fails if + either changes (``_KEY_PATTERNS`` in tests/test_schema_pattern_end_anchor.py). """ - return _end_anchored_validator(jsonschema.validators.validator_for(schema))(schema) + if not JSONSCHEMA_AVAILABLE: + raise ImportError("jsonschema is not available") from _JSONSCHEMA_IMPORT_ERROR + if default is None: + validator_class = jsonschema.validators.validator_for(schema) + else: + validator_class = jsonschema.validators.validator_for(schema, default=default) + return _end_anchored_validator(validator_class)(schema) def _schema_version_sort_key(version: str) -> tuple[int, int, int]: diff --git a/tests/test_schema_pattern_end_anchor.py b/tests/test_schema_pattern_end_anchor.py index 6b12793da..b51d057bc 100644 --- a/tests/test_schema_pattern_end_anchor.py +++ b/tests/test_schema_pattern_end_anchor.py @@ -239,12 +239,33 @@ def _nodes(node: Any) -> Iterator[Dict[str, Any]]: yield from _nodes(value) +#: Every schema fluid validates against: the FLUID and Ossie ones, and the vendored +#: standards (tests/test_standard_schema_pattern_end_anchor.py tests their checks). +_SHIPPED = sorted(_SCHEMA_DIR.glob("*.json")) + [ + _SCHEMA_DIR.parent / "providers" / "odcs" / "odcs-schema-v3.1.0.json", + _SCHEMA_DIR.parent / "providers" / "odps_standard" / "schemas" / "odps-product-v1.0.0.json", + _SCHEMA_DIR.parent / "providers" / "odps_standard" / "schemas" / "odps-product-v1.1.0.json", + _SCHEMA_DIR.parent / "providers" / "opds" / "opds-schema-v4.1.0.json", +] + +#: ``patternProperties`` keys, by schema. They keep jsonschema's own ``re.search``: +#: jsonschema matches them in ``patternProperties``, ``additionalProperties`` and +#: ``unevaluatedProperties``, and overriding ``patternProperties`` alone, as +#: check-jsonschema does, lets a key that one matches and another does not through +#: ``additionalProperties: false`` unchecked. So OPDS's ISO 639-1 language codes, +#: ``^[a-z]{2}$``, also match ``"en\n"``. Under ``product.details`` and ``declarative``, +#: which take any key, the value is then checked as a language entry. +#: ``$defs/Languages`` takes no other key, so a hand-written OPDS file can carry +#: ``{"en\n": ...}`` there; fluid's emitter writes ``en``. +_KEY_PATTERNS = {"opds-schema-v4.1.0.json": [("^[a-z]{2}$",)] * 3} + + def test_every_bundled_schema_pattern_is_rewritten_anchor_for_anchor() -> None: """The scan reads escapes and sets as Python does, checked against Python's own - parser for every ``pattern`` the bundled schemas hold.""" + parser for every ``pattern`` the shipped schemas hold.""" patterns = { node["pattern"] - for path in _SCHEMA_DIR.glob("fluid-*schema-*.json") + for path in _SHIPPED for node in _nodes(json.loads(path.read_text(encoding="utf-8"))) if isinstance(node.get("pattern"), str) } @@ -252,16 +273,17 @@ def test_every_bundled_schema_pattern_is_rewritten_anchor_for_anchor() -> None: assert [p for p in patterns if not _only_end_anchors_changed(p)] == [] -@pytest.mark.parametrize( - "path", sorted(_SCHEMA_DIR.glob("fluid-schema-*.json")), ids=lambda path: path.name -) +@pytest.mark.parametrize("path", _SHIPPED, ids=lambda path: path.name) def test_every_pattern_of_the_schema_is_checked_with_the_rewrite(path: Path) -> None: """jsonschema picks the validator class again for each subschema it enters, by - that subschema's ``$schema``. One that declares it, or a ``$ref`` to the root, - which does, would be checked with jsonschema's own ``pattern`` again. - ``patternProperties`` keys are not rewritten (``schema_validator`` says why), so - a schema that starts to use the keyword has to come back here.""" - below_root = list(_nodes(json.loads(path.read_text(encoding="utf-8"))))[1:] - assert [n for n in below_root if isinstance(n.get("$schema"), str)] == [] - assert [n for n in below_root if n.get("$ref") == "#"] == [] - assert [n for n in below_root if "patternProperties" in n] == [] + that subschema's ``$schema``. One that declares it, or a ``$ref``, + ``$recursiveRef`` or ``$dynamicRef`` to the root, which does, would be checked + with jsonschema's own ``pattern`` again. ``patternProperties`` keys are not + rewritten, so a schema that starts to use the keyword has to come back to + :data:`_KEY_PATTERNS`.""" + nodes = list(_nodes(json.loads(path.read_text(encoding="utf-8")))) + assert [n for n in nodes[1:] if isinstance(n.get("$schema"), str)] == [] + assert [n for n in nodes if n.get("$ref") == "#"] == [] + assert [n for n in nodes if "$recursiveRef" in n or "$dynamicRef" in n] == [] + keys = [tuple(n["patternProperties"]) for n in nodes if "patternProperties" in n] + assert keys == _KEY_PATTERNS.get(path.name, []) diff --git a/tests/test_standard_schema_pattern_end_anchor.py b/tests/test_standard_schema_pattern_end_anchor.py new file mode 100644 index 000000000..23eb38ec3 --- /dev/null +++ b/tests/test_standard_schema_pattern_end_anchor.py @@ -0,0 +1,315 @@ +# 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. + +"""The other schema checks read a ``pattern``'s ``$`` as ECMA-262 does, too. + +``tests/test_schema_pattern_end_anchor.py`` covers the FLUID schemas. fluid also +validates against the vendored ODCS, Bitol ODPS and OPDS schemas, the Ossie +schema, and the schema an installed plugin gives its ``extensions.`` block. +Each of those checks ran jsonschema's own ``pattern``, Python's ``re.search``, +whose ``$`` also matches before a final ``\\n``. So ODCS refused +``schema[].id: "t 1"`` but took ``"t1\\n"``, which an ``id: |`` block produces. + +Of the vendored schemas only ODCS has an end-anchored ``pattern`` today, so the +other checks are tested with a schema of the test's own. The tripwires in +``tests/test_schema_pattern_end_anchor.py`` cover the vendored schemas too: what +they would have to start using for a check to leave the rewrite. +""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path +from typing import Any, Dict, Iterator + +import jsonschema +import pytest +import yaml + +import fluid_build +import fluid_build.extension_schemas as extension_schemas +from fluid_build.forge.core.artifact_validators import validate_odcs, validate_opds +from fluid_build.forge_datamodel.emit import osi_sidecar +from fluid_build.providers.base import ProviderError +from fluid_build.providers.odcs.validation import collect_errors, load_schema +from fluid_build.providers.odps_standard import validation as odps_validation +from fluid_build.providers.opds import validator as opds_validator +from tests.test_jsonschema_dialect import _ODCS_BASE + +pytestmark = [pytest.mark.unit] + +_LINE_BREAK_HINT = " (it ends in a line break, which a YAML | or > block keeps and |- or >- drops)" +#: A schema of the test's own, for the checks whose vendored schema has no pattern. +_NAME_SCHEMA: Dict[str, Any] = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": {"name": {"type": "string", "pattern": "^[a-z]+$"}}, +} + + +def _odcs(table_id: str) -> Dict[str, Any]: + return {**_ODCS_BASE, "schema": [{"id": table_id, "name": "orders", "logicalType": "object"}]} + + +def _odcs_yaml(block: str, table_id: str = "t1") -> str: + """An ODCS document whose ``schema[0].id`` is a YAML block scalar: ``|`` or ``|-``.""" + body = yaml.safe_dump(_ODCS_BASE) + return f"{body}schema:\n- id: {block}\n {table_id}\n name: orders\n logicalType: object\n" + + +# ── ODCS: fluid odcs validate, fluid validate-artifacts ───────────────────── + + +def test_odcs_refuses_a_table_id_that_ends_in_a_newline() -> None: + """``collect_errors`` is ``fluid odcs validate``'s jsonschema pass.""" + schema = load_schema() + assert schema is not None + assert collect_errors(_odcs("t1"), schema) == [] + + errors = [e for e in collect_errors(_odcs("t1\n"), schema) if e["validator"] == "pattern"] + + assert [e["path"] for e in errors] == ["schema.0.id"], errors + assert errors[0]["message"].endswith(_LINE_BREAK_HINT), errors + + +def test_fluid_odcs_validate_refuses_a_block_scalar_table_id(tmp_path: Path, capsys: Any) -> None: + from fluid_build.cli import main + + path = tmp_path / "orders.odcs.yaml" + path.write_text(_odcs_yaml("|"), encoding="utf-8") + assert yaml.safe_load(path.read_text(encoding="utf-8"))["schema"][0]["id"] == "t1\n" + + rc = main(["odcs", "validate", str(path), "--no-vowl"]) + + text = capsys.readouterr().out + assert rc == 1, text + assert "schema.0.id: 't1\\n' does not match" in text, text + + +def test_fluid_odcs_validate_keeps_the_hint_whole(tmp_path: Path, capsys: Any) -> None: + """The command cuts each message to 120 characters. Cut inside the hint, a long id's + message ended ``... block keeps and |``, which reads as advice to use ``|``.""" + from fluid_build.cli import main + + path = tmp_path / "orders.odcs.yaml" + path.write_text(_odcs_yaml("|", "customer_orders_v1"), encoding="utf-8") + + rc = main(["odcs", "validate", str(path), "--no-vowl"]) + + text = " ".join(capsys.readouterr().out.split()) # undo the console's line wrapping + assert rc == 1, text + assert f"'customer_orders_v1\\n' does not match '^[A-Za-z0-9_-]+$'{_LINE_BREAK_HINT}" in text + + +def test_fluid_odcs_validate_prints_a_value_with_brackets_literally( + tmp_path: Path, capsys: Any +) -> None: + """The message quotes the value, and Rich read ``[/]`` in it as a closing tag: the + command ended in ``Unexpected error: closing tag '[/]' ... has nothing to close``.""" + from fluid_build.cli import main + + path = tmp_path / "orders.odcs.yaml" + path.write_text(yaml.safe_dump(_odcs("t[/]1")), encoding="utf-8") + + rc = main(["odcs", "validate", str(path), "--no-vowl"]) + + text = " ".join(capsys.readouterr().out.split()) + assert rc == 1, text + assert "schema.0.id: 't[/]1' does not match" in text, text + + +def test_validate_artifacts_refuses_a_block_scalar_table_id() -> None: + """``validate_odcs`` is what ``fluid validate-artifacts`` runs on ``odcs/*.odcs.yaml``.""" + assert validate_odcs("odcs/orders.odcs.yaml", _odcs_yaml("|-").encode()) == [] + + issues = validate_odcs("odcs/orders.odcs.yaml", _odcs_yaml("|").encode()) + + messages = [i.message for i in issues] + assert any(m.startswith("at schema/0/id: 't1\\n' does not match ") for m in messages), messages + + +# ── Bitol ODPS, OPDS, Ossie: no pattern in the vendored schema yet ────────── + + +def test_odps_validate_reads_a_pattern_as_ecma_does() -> None: + odps_validation.validate({"name": "orders"}, _NAME_SCHEMA) + + with pytest.raises(ProviderError, match=re.escape("'orders\\n' does not match '^[a-z]+$'")): + odps_validation.validate({"name": "orders\n"}, _NAME_SCHEMA) + + +def test_odps_validate_still_checks_the_schema_first() -> None: + """It still runs ``jsonschema.validate``, now with the rewritten class.""" + with pytest.raises(jsonschema.SchemaError): + odps_validation.validate({}, {"type": 5}) + + +def test_opds_validation_reads_a_pattern_as_ecma_does(monkeypatch: Any) -> None: + monkeypatch.setitem(opds_validator._SCHEMA_CACHE, "0.0-test", _NAME_SCHEMA) + validate = opds_validator.validate_against_opds_schema + assert validate({"name": "orders"}, "0.0-test") == (True, None, "full_schema") + + valid, errors, kind = validate({"name": "orders\n"}, "0.0-test") + + assert (valid, kind) == (False, "full_schema") + assert errors == [f"name: 'orders\\n' does not match '^[a-z]+$'{_LINE_BREAK_HINT}"] + + +def test_an_opds_schema_that_declares_no_dialect_is_2020_12(monkeypatch: Any) -> None: + """What the check pinned before. Draft 7 has no ``dependentRequired`` and would + pass this document.""" + schema = {"type": "object", "dependentRequired": {"a": ["b"]}} + monkeypatch.setitem(opds_validator._SCHEMA_CACHE, "0.0-test", schema) + + valid, errors, _ = opds_validator.validate_against_opds_schema({"a": 1}, "0.0-test") + + assert (valid, errors) == (False, ["root: 'b' is a dependency of 'a'"]) + + +def test_ossie_document_validation_reads_a_pattern_as_ecma_does(monkeypatch: Any) -> None: + monkeypatch.setattr(osi_sidecar, "_load_ossie_schema", lambda: _NAME_SCHEMA) + assert osi_sidecar.validate_osi_document({"name": "orders"}) == [] + + issues = osi_sidecar.validate_osi_document({"name": "orders\n"}) + + assert issues == [f"name: 'orders\\n' does not match '^[a-z]+$'{_LINE_BREAK_HINT}"] + + +def test_without_jsonschema_each_check_says_it_is_missing(monkeypatch: Any) -> None: + """The checks read the flag ``schema_validator`` needs. A jsonschema that is found + but fails to import is reported as missing, not as ``NameError: jsonschema``.""" + monkeypatch.setattr("fluid_build.schema_manager.JSONSCHEMA_AVAILABLE", False) + + assert [i.code for i in validate_odcs("odcs/o.odcs.yaml", b"id: x")] == ["ODCS-LIB-MISSING"] + assert [i.code for i in validate_opds("opds/o.opds.json", b"{}")] == ["OPDS-LIB-MISSING"] + assert collect_errors(_odcs("t1\n"), {}) == [] + with pytest.raises(ImportError, match="jsonschema is not available"): + osi_sidecar.validate_osi_document({}) + + +# ── plugin extension blocks ───────────────────────────────────────────────── + +#: The shape of data-product-forge-custom-scaffold's schema: draft-07, "$"-anchored. +_SCAFFOLD_SCHEMA: Dict[str, Any] = { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "patterns": { + "type": "array", + "items": { + "type": "object", + "properties": { + "use": { + "type": "string", + "pattern": "^[a-zA-Z][a-zA-Z0-9_-]*:[a-zA-Z][a-zA-Z0-9_-]*$", + } + }, + }, + } + }, +} + + +def _assemble(monkeypatch: Any, schema: Dict[str, Any], block: Any) -> Dict[str, Any]: + monkeypatch.setattr(extension_schemas, "iter_extension_schemas", lambda *a, **k: {"x": schema}) + return extension_schemas.assemble_proposed_extensions({"x": block}) + + +def test_an_extension_block_whose_value_ends_in_a_newline_is_dropped(monkeypatch: Any) -> None: + kept = {"patterns": [{"use": "ci:basic"}]} + assert _assemble(monkeypatch, _SCAFFOLD_SCHEMA, kept) == {"x": kept} + assert _assemble(monkeypatch, _SCAFFOLD_SCHEMA, {"patterns": [{"use": "ci:basic\n"}]}) == {} + + +def test_an_extension_schema_is_read_in_the_dialect_it_declares(monkeypatch: Any) -> None: + """Draft 7 ignores ``unevaluatedProperties``, so the extra key used to be kept.""" + schema = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": {"libraries": {"type": "array"}}, + "unevaluatedProperties": False, + } + assert _assemble(monkeypatch, schema, {"libraries": []}) == {"x": {"libraries": []}} + assert _assemble(monkeypatch, schema, {"libraries": [], "librarys": []}) == {} + + +def test_an_extension_schema_that_cannot_be_applied_drops_only_its_block( + monkeypatch: Any, caplog: Any +) -> None: + """A malformed plugin schema neither stops the run nor lets its block through.""" + schemas = {"bad": {"$schema": 7, "type": "string"}, "x": _SCAFFOLD_SCHEMA} + monkeypatch.setattr(extension_schemas, "iter_extension_schemas", lambda *a, **k: schemas) + kept = {"patterns": [{"use": "ci:basic"}]} + + assert extension_schemas.assemble_proposed_extensions({"bad": "a", "x": kept}) == {"x": kept} + assert "extension schema 'bad' unusable" in caplog.text + + +def test_an_extension_schema_that_declares_no_dialect_is_draft_07(monkeypatch: Any) -> None: + """The plugin contract says draft-07. 2020-12, jsonschema's default, has no + ``dependencies`` keyword and would keep this block.""" + schema = {"type": "object", "dependencies": {"output": ["patterns"]}} + kept = {"output": "x", "patterns": []} + assert _assemble(monkeypatch, schema, kept) == {"x": kept} + assert _assemble(monkeypatch, schema, {"output": "x"}) == {} + + +# ── one way to build a validator ──────────────────────────────────────────── + + +def _validator_builds(path: Path) -> Iterator[str]: + """Spellings in ``path`` that build a jsonschema validator without ``schema_validator``. + It goes by names, so a class first bound to another name (``V = Draft7Validator``) + gets past it.""" + text = path.read_text(encoding="utf-8") + if not any(word in text for word in ("jsonschema", "Validator", "validator_for")): + return + tree = ast.parse(text) + modules = {"jsonschema"} # names that hold jsonschema or jsonschema.validators + for node in ast.walk(tree): + if isinstance(node, ast.Import): + modules |= {a.asname for a in node.names if a.name == "jsonschema" and a.asname} + elif isinstance(node, ast.ImportFrom) and (node.module or "").startswith("jsonschema"): + modules |= {a.asname or a.name for a in node.names if a.name == "validators"} + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and (node.module or "").startswith("jsonschema"): + for a in node.names: + builder = a.name == "validator_for" or re.fullmatch(r"Draft\d+Validator", a.name) + if a.name == "validate" or (a.asname and builder): + yield f"{path.name}:{node.lineno}: from {node.module} import {a.name}" + if not isinstance(node, ast.Call): + continue + func = node.func + name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") + if name == "validator_for" or re.fullmatch(r"Draft\d+Validator", name): + yield f"{path.name}:{node.lineno}: {name}()" + elif name in ("validate", "extend", "create") and isinstance(func, ast.Attribute): + root = func.value + while isinstance(root, ast.Attribute): + root = root.value + cls = next((k.value for k in node.keywords if k.arg == "cls"), None) + routed = cls is not None and "schema_validator" in ast.unparse(cls) + if isinstance(root, ast.Name) and root.id in modules and not routed: + yield f"{path.name}:{node.lineno}: {ast.unparse(func)}()" + + +def test_every_validator_is_built_by_schema_validator() -> None: + """A check that builds its own validator reads ``$`` as Python does again; #799 and + this change each found such checks one at a time. Give ``schema_validator`` the + dialect class as ``default=`` instead of building it.""" + package = Path(fluid_build.__file__).parent + paths = [p for p in sorted(package.rglob("*.py")) if p != package / "schema_manager.py"] + assert len(paths) > 500, len(paths) # the glob found the package + assert [build for path in paths for build in _validator_builds(path)] == [] From 8c8116ed828c63f23ad9d4aded1294919c34c670 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 4 Nov 2026 05:36:40 +0100 Subject: [PATCH 3/6] feat(validate,plan): warn about an id the build phase refuses (#813) * feat(validate,plan): warn about an id the build phase refuses The schema's $defs.identifier admits ids the build phase refuses: a leading digit (1orders) or more than 128 characters. fluid apply in a build mode refuses such an id before it applies anything (#803), but fluid validate and fluid plan said nothing, so the first sign was a failed apply. The two grammars stay different by decision. For a contract with builds, both now warn and name the id, using the rule apply uses (build_runners._ids.validate_build_ids): - validate adds a ValidationResult warning, so plain validate passes and --strict fails (founder's call). It runs with the other 0.7.x contract rules, on a contract file and on a bundle's resolved contract. If the check itself crashes, that is reported as a warning. - plan prints the warning and still writes plan.json, whatever its --mode (founder's call). It runs after the gate plan shares with apply, not inside it, so apply keeps its typed CLIError. IDENT_RULE states the grammar in words, and both the new warning and apply's hint use it. The hint said validate and plan "accept" such an id; it now says they only warn about it. * fix(validate,plan): keep the build-id warning true on a schema-refused id Review follow-ups to the build-id warning: - validate runs its contract rules on a contract the schema refused too, so the warning could say "The contract schema allows more" about an id the schema had just refused (id: 123). The text now only says what the build phase needs and that fluid apply in a build mode refuses the id. - plan prints the warning the way validate prints its warnings: literal, soft_wrap (the terminal wraps, so Rich never breaks a line inside the id) and no highlighter. --- CHANGELOG.md | 12 + fluid_build/build_runners/_ids.py | 30 +++ fluid_build/cli/apply.py | 15 +- fluid_build/cli/plan.py | 24 +- fluid_build/cli/validate.py | 29 +++ .../test_validate_plan_build_ids_warning.py | 229 ++++++++++++++++++ 6 files changed, 332 insertions(+), 7 deletions(-) create mode 100644 tests/cli/test_validate_plan_build_ids_warning.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c27efb8f..c87996c7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`fluid validate` and `fluid plan` warn about an id the build phase refuses.** The build + phase puts `contract.id` and each `builds[].id` in directory names and generated code, so they + must start with a letter or underscore, use only letters, digits, `_`, `.` and `-`, and be at + most 128 characters. The contract schema allows more (`1orders`, or a longer id), and + `fluid apply` in a build mode refuses such an id before it applies anything, but validate and + plan said nothing. For a contract with builds, both now warn and name the id: plain + `fluid validate` passes and `--strict` fails, and `fluid plan` still writes the plan, whatever + its `--mode`. A contract without builds is not checked, as in `fluid apply`. + - **IaC plugins load when a contract needs their platform, and cannot take a built-in's name unless the operator says so.** A package that ships an OpenTofu emitter through the `fluid_build.iac_providers` entry-point group used to be imported whenever `fluid_build.iac` @@ -569,6 +578,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Upgrade notes +- New validate warning, so `--strict` fails: a contract with builds whose `contract.id` or a + `builds[].id` the build phase refuses (`1orders`, or more than 128 characters). `fluid apply` + in a build mode already refuses it; rename the id to fit. - A contract with a column typed `array`, `struct`, `map`, `object`, `record` or `row` (or `array(...)`) on an `aws` Glue table or a `gcp` BigQuery table now fails `fluid validate`, `fluid generate iac` and `fluid apply` with `column-type`, where it validated and wrote a diff --git a/fluid_build/build_runners/_ids.py b/fluid_build/build_runners/_ids.py index 5fdb0f241..c68e89c89 100644 --- a/fluid_build/build_runners/_ids.py +++ b/fluid_build/build_runners/_ids.py @@ -47,6 +47,12 @@ # can never traverse out of the directory it is joined into. _IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.\-]{0,127}\Z") +# :data:`_IDENT_RE` in words, for a message that tells an author what to change. +IDENT_RULE = ( + "start with a letter or underscore, use only letters, digits, '_', '.' and '-', " + "and be at most 128 characters" +) + class IdentifierViolation(ValueError): """Raised when a contract identifier fails validation. @@ -103,3 +109,27 @@ def validate_build_ids(contract: Mapping[str, Any]) -> None: for build in builds: if isinstance(build, Mapping) and build.get("id"): validate_identifier(build["id"], kind="build.id") + + +def build_id_warning(contract: Mapping[str, Any]) -> Optional[str]: + """The warning ``fluid validate`` and ``fluid plan`` give for an id the build phase refuses. + + The schema's ``$defs.identifier`` admits more than :data:`_IDENT_RE` (a + leading digit, any length), and the two stay different by decision. So + validate and plan accept such an id and warn, and ``fluid apply`` in a + build mode refuses it. The check is :func:`validate_build_ids`, so the + warning names the id apply would name. Returns ``None`` when it passes. + + The text does not say the schema accepted the id: validate also runs this + on a contract the schema refused, such as one whose ``id`` is ``123``. + """ + try: + validate_build_ids(contract) + except IdentifierViolation as exc: + return ( + f"{exc.kind} {exc.value!r} cannot be used by the build phase, which puts it in " + f"directory names and generated code, so it must {IDENT_RULE}. fluid apply in a " + "build mode (amend-and-build, replace-and-build) refuses this id before it " + "applies anything. Rename it to fit." + ) + return None diff --git a/fluid_build/cli/apply.py b/fluid_build/cli/apply.py index c702ea26e..ce988ae99 100644 --- a/fluid_build/cli/apply.py +++ b/fluid_build/cli/apply.py @@ -313,7 +313,8 @@ def _gate_build_ids_for_apply(contract: Dict[str, Any]) -> None: The build phase turns ``contract.id`` and ``builds[].id`` into directory names, so it takes a narrower grammar than the schema: ``1orders``, or an - id of 129 characters, passes ``fluid validate`` and ``fluid plan``. The + id of 129 characters, passes ``fluid validate`` and ``fluid plan``, which + only warn (``build_runners._ids.build_id_warning``). The build runner checks the ids itself, but on the OpenTofu engine it runs after ``tofu apply``, so such an id was refused once the product was provisioned, as an "Unexpected error". This is the runner's own check @@ -323,7 +324,11 @@ def _gate_build_ids_for_apply(contract: Dict[str, Any]) -> None: Raises: CLIError: ``apply_build_identifier_invalid``. """ - from fluid_build.build_runners._ids import IdentifierViolation, validate_build_ids + from fluid_build.build_runners._ids import ( + IDENT_RULE, + IdentifierViolation, + validate_build_ids, + ) try: validate_build_ids(contract) @@ -337,10 +342,8 @@ def _gate_build_ids_for_apply(contract: Dict[str, Any]) -> None: "hint": ( "The build phase keeps each build's run records, cursors and " "watermarks under .fluid/runs///, so both ids " - "must start with a letter or underscore, use only letters, digits, " - "'_', '.' and '-', and be at most 128 characters. The contract schema " - "allows more, so fluid validate and fluid plan accept it. Nothing was " - "applied." + f"must {IDENT_RULE}. The contract schema allows more, so fluid " + "validate and fluid plan only warn about it. Nothing was applied." ), }, ) from exc diff --git a/fluid_build/cli/plan.py b/fluid_build/cli/plan.py index 3faab1f48..456321ecc 100644 --- a/fluid_build/cli/plan.py +++ b/fluid_build/cli/plan.py @@ -30,7 +30,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional -from fluid_build.cli.console import cprint, warning +from fluid_build.cli.console import cprint, cprint_literal, warning # 11-stage pipeline: stage-6 → stage-7 cryptographic plan-binding. ``plan.json`` # carries ``bundleDigest`` (pins input bundle) + ``planDigest`` (catches @@ -305,6 +305,7 @@ def run(args, logger: logging.Logger) -> int: # plan generation or write. A failure raises ``CLIError`` → # non-zero exit, no plan.json written. _gate_contract_for_plan_or_apply(contract, logger, command="plan") + _warn_unusable_build_ids(contract) # --- --provider gate ------------------------------------------------- # ``--provider`` reaches ``build_provider`` only on the legacy planner @@ -761,6 +762,27 @@ def _gate_contract_for_plan_or_apply( ) +def _warn_unusable_build_ids(contract: Dict[str, Any]) -> None: + """Warn when the build phase will refuse ``contract.id`` or a ``builds[].id``. + + The schema admits ids the build phase refuses (``1orders``, or one of 129 + characters), and ``fluid apply`` in a build mode refuses such an id before + it applies anything. Plan only warns, whatever ``--mode`` it plans for, and + writes the plan. Not part of :func:`_gate_contract_for_plan_or_apply`: + apply shares that gate and refuses the id with a typed error of its own + (``cli/apply.py::_gate_build_ids_for_apply``). Printed, not logged: a + ``warn()`` event would print the same text again as a JSON line. + """ + from fluid_build.build_runners._ids import build_id_warning + + message = build_id_warning(contract) + if message is not None: + # As validate prints its warnings (validate._cprint_literal): literal, + # because the id is contract data, not markup; the terminal wraps the + # line, so Rich never breaks one inside the id; and no highlighter. + cprint_literal(f"\n⚠️ {message}", highlight=False, soft_wrap=True) + + def _validate_plan_actions(plan: Dict[str, Any], logger: logging.Logger) -> None: """Run SDK ``validate_actions()`` over the plan's action list. diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py index 0183b04e2..c47696400 100644 --- a/fluid_build/cli/validate.py +++ b/fluid_build/cli/validate.py @@ -1186,6 +1186,9 @@ def _run_contract_rules( "the contract was NOT checked for a continuous trigger on a batch build" ) + # --- An id the build phase refuses (WARNING) ----------------------- + _build_id_check(validation_result, contract) + # --- Packaging checks (RFC-packaging-modes.md file 9) --------------- # ``fluid validate`` had NO packaging awareness: a `shared` block with # no `pool` validated clean and then failed at `fluid plan` with an @@ -1306,6 +1309,32 @@ def _databricks_binding_check(validation_result, contract: Mapping[str, Any]) -> validation_result.add_warning(msg) +def _build_id_check(validation_result, contract: Mapping[str, Any]) -> None: + """A ``contract.id`` or ``builds[].id`` the build phase refuses, as a warning. + + The schema's ``$defs.identifier`` admits ids the build phase refuses + (``1orders``, or more than 128 characters), and the two grammars stay + different by decision. ``fluid apply`` in a build mode refuses such an id + before it applies anything, with the same rule + (``build_runners._ids.validate_build_ids``). A warning, not an error: plain + validate passes and ``--strict`` fails. A crash in the check is reported, + never swallowed. + """ + try: + from fluid_build.build_runners._ids import build_id_warning + + message = build_id_warning(contract) + except Exception as exc: # noqa: BLE001 — reported, never swallowed + validation_result.add_warning( + f"Build-id check could not run ({type(exc).__name__}: {exc}); " + "contract.id and builds[].id were NOT checked against the ids the build " + "phase accepts" + ) + return + if message is not None: + validation_result.add_warning(message) + + def _duplicate_contract_id_errors(contract: Mapping[str, Any]) -> List[str]: """Refuse two exposes whose output ports would be published under one contract id. diff --git a/tests/cli/test_validate_plan_build_ids_warning.py b/tests/cli/test_validate_plan_build_ids_warning.py new file mode 100644 index 000000000..5fb393bf3 --- /dev/null +++ b/tests/cli/test_validate_plan_build_ids_warning.py @@ -0,0 +1,229 @@ +# 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. + +"""``fluid validate`` and ``fluid plan`` warn about an id the build phase refuses. + +The schema's ``$defs.identifier`` admits ids the build phase refuses (``1orders``, +or an id longer than 128 characters), and the two grammars stay different by +decision. ``fluid apply`` in a build mode refuses such an id before it applies +anything (``test_apply_build_ids_before_tofu.py``). Validate and plan used to say +nothing about it. Now validate warns, so ``--strict`` fails, and plan warns and +still writes the plan, whatever its ``--mode``. The warning comes from the rule +apply uses, ``build_runners._ids.validate_build_ids``, so the three agree on +which ids are refused. +""" + +from __future__ import annotations + +import copy +import json +import logging +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Dict + +import pytest +import yaml + +import fluid_build.cli as cli_mod +from fluid_build.build_runners import _ids +from fluid_build.cli.plan import _gate_contract_for_plan_or_apply, _warn_unusable_build_ids +from fluid_build.cli.plan import run as plan_run +from fluid_build.cli.validate import _run_contract_rules +from fluid_build.cli.validate import run as validate_run +from fluid_build.schema_manager import ValidationResult +from tests.cli.test_apply_build_ids_before_tofu import _CONTRACT, _UNUSABLE, _with + +LOG = logging.getLogger("test.validate_plan_build_ids_warning") + +# What the warning says besides the id, in the words an author acts on. +_RULE = "start with a letter or underscore" +_REFUSED_BY = "fluid apply in a build mode" + + +def _squash(text: str) -> str: + """``text`` without whitespace: Rich folds a long id across lines.""" + return "".join(text.split()) + + +def _names(out: str, field: str, value: str) -> bool: + return _squash(f"{field} {value!r}") in _squash(out) + + +def _without_builds(contract: Dict[str, Any]) -> Dict[str, Any]: + contract = copy.deepcopy(contract) + del contract["builds"] + return contract + + +@pytest.fixture +def workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("FLUID_ENV", raising=False) + return tmp_path + + +def _write(contract: Dict[str, Any]) -> str: + path = Path("contract.fluid.yaml") + path.write_text(yaml.safe_dump(contract, sort_keys=False), encoding="utf-8") + return str(path) + + +def _validate(contract: Dict[str, Any], *flags: str) -> int: + """``fluid validate contract.fluid.yaml --offline `` through the real parser.""" + argv = ["validate", _write(contract), "--offline", *flags] + return validate_run(cli_mod.build_parser().parse_args(argv), LOG) + + +def _plan(contract: Dict[str, Any], *flags: str) -> int: + """``fluid plan contract.fluid.yaml --out plan.json `` through the real parser.""" + argv = ["plan", _write(contract), "--out", "plan.json", *flags] + return plan_run(cli_mod.build_parser().parse_args(argv), LOG) + + +class TestTheWarning: + @pytest.mark.parametrize("field,value", _UNUSABLE) + def test_it_names_the_id_the_rule_and_the_refusal(self, field: str, value: str) -> None: + message = _ids.build_id_warning(_with(field, value)) + assert message is not None + assert message.startswith(f"{field} {value!r} ") + assert _RULE in message + assert "'_', '.' and '-'" in message + assert "at most 128 characters" in message + assert _REFUSED_BY in message + + def test_usable_ids_get_none(self) -> None: + assert _ids.build_id_warning(copy.deepcopy(_CONTRACT)) is None + + @pytest.mark.parametrize( + "contract", + [ + pytest.param(_without_builds(_with("contract.id", "1orders")), id="no-builds"), + pytest.param({"id": "1orders", "builds": []}, id="empty-builds"), + pytest.param({"id": "", "builds": [{"id": ""}]}, id="empty-ids"), + pytest.param({"builds": [{"engine": "sql"}]}, id="absent-ids"), + pytest.param({"id": "orders", "builds": ["1make_rows", None]}, id="non-mapping"), + ], + ) + def test_what_apply_does_not_check_gets_none(self, contract: Dict[str, Any]) -> None: + """The ids ``validate_build_ids`` skips are not warned about either.""" + assert _ids.build_id_warning(contract) is None + + +@pytest.mark.usefixtures("workspace") +class TestValidate: + @pytest.mark.parametrize("field,value", _UNUSABLE) + def test_plain_validate_warns_and_passes( + self, capsys: pytest.CaptureFixture[str], field: str, value: str + ) -> None: + assert _validate(_with(field, value)) == 0 + out = capsys.readouterr().out + assert _names(out, field, value) + assert _squash(_REFUSED_BY) in _squash(out) + + @pytest.mark.parametrize("field,value", _UNUSABLE) + def test_strict_fails(self, capsys: pytest.CaptureFixture[str], field: str, value: str) -> None: + assert _validate(_with(field, value), "--strict") == 1 + out = " ".join(capsys.readouterr().out.split()) + assert "warnings treated as errors (--strict)" in out + + def test_json_lists_the_warning(self, capsys: pytest.CaptureFixture[str]) -> None: + assert _validate(_with("contract.id", "1orders"), "--format", "json") == 0 + report = json.loads(capsys.readouterr().out) + assert report["valid"] is True + assert [w for w in report["warnings"] if w.startswith("contract.id '1orders' ")] + + def test_usable_ids_pass_strict(self) -> None: + """The base contract is clean, so the warning is what fails ``--strict`` above.""" + assert _validate(copy.deepcopy(_CONTRACT), "--strict") == 0 + + def test_a_contract_without_builds_is_not_warned(self) -> None: + assert _validate(_without_builds(_with("contract.id", "1orders")), "--strict") == 0 + + def test_a_schema_refused_id_is_not_called_schema_valid( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + """Contract rules also run on a contract the schema refused, so the warning + must not claim the schema accepted the id.""" + assert _validate(_with("contract.id", 123)) == 1 + out = capsys.readouterr().out + assert _squash("contract.id 123 cannot be used by the build phase") in _squash(out) + assert "schema allows" not in " ".join(out.split()) + + def test_a_crash_in_the_check_is_reported(self, monkeypatch: pytest.MonkeyPatch) -> None: + def boom(contract: Any) -> None: + raise RuntimeError("boom") + + monkeypatch.setattr(_ids, "build_id_warning", boom) + result = ValidationResult(is_valid=True) + args = SimpleNamespace(verbose=False, quiet=True, strict=False, offline=True) + _run_contract_rules(copy.deepcopy(_CONTRACT), None, result, args, LOG) + assert [w for w in result.warnings if w.startswith("Build-id check could not run")] + + +@pytest.mark.usefixtures("workspace") +class TestPlan: + @pytest.mark.parametrize("field,value", _UNUSABLE) + def test_plan_warns_and_writes_the_plan( + self, capsys: pytest.CaptureFixture[str], field: str, value: str + ) -> None: + assert _plan(_with(field, value)) == 0 + assert Path("plan.json").is_file() + out = capsys.readouterr().out + assert _names(out, field, value) + assert _squash(_REFUSED_BY) in _squash(out) + + @pytest.mark.parametrize("mode", ["amend", "amend-and-build", "replace-and-build"]) + def test_every_mode_only_warns(self, capsys: pytest.CaptureFixture[str], mode: str) -> None: + """Even the build modes whose apply will refuse it (founder's call).""" + assert _plan(_with("contract.id", "1orders"), "--mode", mode) == 0 + assert json.loads(Path("plan.json").read_text(encoding="utf-8"))["mode"] == mode + assert _names(capsys.readouterr().out, "contract.id", "1orders") + + def test_it_is_printed_once( + self, capsys: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture + ) -> None: + """Not also as a log event, which the console prints again as a JSON line.""" + with caplog.at_level(logging.DEBUG): + assert _plan(_with("build.id", "1make_rows")) == 0 + captured = capsys.readouterr() + assert _squash(captured.out + captured.err).count(_squash(_REFUSED_BY)) == 1 + assert not [r for r in caplog.records if _REFUSED_BY in r.getMessage()] + + def test_a_marked_up_id_is_printed_as_written(self, capsys: pytest.CaptureFixture[str]) -> None: + """The id is contract data: Rich must not read ``[/]`` as a tag. The schema + refuses ``[`` today, so this calls the warning past the gate.""" + _warn_unusable_build_ids(_with("build.id", "[/]make_rows")) + assert _names(capsys.readouterr().out, "build.id", "[/]make_rows") + + def test_usable_ids_are_not_warned(self, capsys: pytest.CaptureFixture[str]) -> None: + assert _plan(copy.deepcopy(_CONTRACT)) == 0 + assert _squash(_REFUSED_BY) not in _squash(capsys.readouterr().out) + + def test_a_contract_without_builds_is_not_warned( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + assert _plan(_without_builds(_with("contract.id", "1orders"))) == 0 + assert _squash(_REFUSED_BY) not in _squash(capsys.readouterr().out) + + +def test_the_gate_apply_shares_with_plan_does_not_warn( + capsys: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture +) -> None: + """``fluid apply`` refuses the id in a build mode (``CLIError``) and ignores it in + the others, so the warning stays out of the gate plan shares with apply.""" + with caplog.at_level(logging.DEBUG, logger=LOG.name): + _gate_contract_for_plan_or_apply(_with("contract.id", "1orders"), LOG, command="apply") + assert _squash(_REFUSED_BY) not in _squash(capsys.readouterr().out) + assert not [r for r in caplog.records if "build_identifier" in r.getMessage()] From 24f1773601b406b49187c464fa01a00030e60f99 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 4 Nov 2026 05:36:46 +0100 Subject: [PATCH 4/6] fix(diff): harden the column type comparison behind --fail-on-breaking (#802) * fix(diff): compare column types strictly enough for --fail-on-breaking Now that fluid diff --baseline reads 0.7.x columns, its type comparison decides --fail-on-breaking for those contracts, and several type changes that break consumers passed as widenings or as info. - Fold case for ASCII type names only. Text the comparison cannot order (non-ASCII, or parameters that are not plain integers, such as enum('a','b') or varchar(max)) is its own type, so it no longer reads as unchanged, as "no parameters" or as a whitespace difference. The finding says so when a non-ASCII name is the reason. - A parameter change that is neither a pure widening nor a pure narrowing is breaking (it was info): varchar -> varchar(10), decimal(10,2) -> decimal(12,1), decimal(10,2) -> decimal. - A bigger parameter widens only for types whose parameter measures capacity (lengths, precisions, the decimal family), not for ones that name something, such as geometry(srid). - decimal, numeric and number widen only when the scale and the integer digits (precision - scale) both hold. - A change of base type widens only when the new type declares no parameters, so int -> decimal(5,0) is breaking. - time -> timestamp, float -> decimal/numeric and int -> float leave the widening table. - The type grammar is ASCII only and is matched in linear time. Prior art: the Apache Iceberg table spec ("Schema Evolution": int -> long, float -> double, date -> timestamp from v3, decimal(P,S) -> decimal(P',S) for P' > P) and Delta Lake "Type widening" (decimal(p,s) -> decimal(p+k1,s+k2) for k1 >= k2 >= 0). * fix(diff): report a name the new contract declares twice The comparison indexes exposes, columns and nested fields by name and keeps the last of a repeated name, so an earlier declaration was never compared against the baseline. An expose id, column name or nested field name that the new contract declares more than once is now a breaking change (expose_duplicate_id, column_duplicate_name, nested_field_duplicate_name), in an expose that is new as well as in one both contracts have, and whatever the baseline holds: compare_contracts is idempotent except for a contract that declares a name twice. The declarations are in the finding's "after". * test(diff): diff the schema-invalid fixtures with validate=False fluid diff --baseline validates both contracts since #805, so a fixture that is not a valid contract (the flat-layout and malformed-schema ones) is diffed with validate=False, as the neighbouring tests are. --- CHANGELOG.md | 38 +++ fluid_build/api/changelog.py | 9 +- fluid_build/api/changelog_rules.py | 310 ++++++++++++++++++--- tests/test_diff_changelog.py | 428 +++++++++++++++++++++++++++-- 4 files changed, 724 insertions(+), 61 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c87996c7e..aa0a66041 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -342,6 +342,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 does, and reports its columns at `...schema.fields.`. No bundled schema accepts the wrapper, so `fluid diff --baseline` refuses such a contract as invalid (see Security). +- **`fluid diff --baseline` compares column types more strictly, so `--fail-on-breaking` stops + passing type changes that break consumers.** Contracts that used to diff clean can now report + breaking type changes (a pipeline that runs `fluid diff --baseline --fail-on-breaking` can start + exiting 1), and some findings that were `info` or `non_breaking` are now `breaking`. Rule by rule: + - A type name that is not plain ASCII is compared as written, not case-folded, so apart from + surrounding whitespace it matches nothing but itself, and the finding says so. Type text the + comparison cannot order (non-ASCII text, or parameters that are not plain integers, such as + `enum('a','b')` or `varchar(max)`) is a different type when the text differs after ASCII case + folding; it used to read as "no parameters" or as a whitespace difference. Case inside a quoted + parameter is still folded, so `enum('a')` and `enum('A')` still compare equal. + - A parameter change that is neither a pure widening nor a pure narrowing is now `breaking`, not + `info`: `varchar` to `varchar(10)` or back, `decimal(10,2)` to `decimal(12,1)` or to a bare + `decimal`. The kind stays `column_type_param_changed`; its text now reads "(not a pure + widening)". + - A bigger parameter is a widening only for types whose parameter measures capacity: lengths + (`varchar`, `char`, `string`, `text`, `varbinary` and their variants), precisions (`float`, + `time`, `timestamp`, `datetime` and their variants) and the decimal family. A parameter that + names something is not one: `geometry(4326)` to `geometry(32633)`, `binary(16)` to + `binary(32)` and `vector(3)` to `vector(1536)` are now `breaking`. + - A `decimal`, `dec`, `numeric`, `number` or `bignumeric` with a precision and a scale widens only + when the scale and the integer digits (precision minus scale) both hold. `decimal(10,2)` to + `decimal(12,4)` is still a widening; `decimal(10,2)` to `decimal(11,4)`, which holds one + integer digit fewer, is now `breaking`. + - A change to a different base type widens only when the new type declares no parameters of its + own: `int` to `decimal(5,0)` is now `breaking`, `int` to a bare `decimal` still widens. + - `time` to `timestamp` or `datetime`, `float` to `decimal` or `numeric`, and `int`, `integer` + or `int32` to `float` are no longer widenings. `float` to `double` and `int` to `double` are. + +- **`fluid diff --baseline` reports a name the new contract declares twice.** The comparison matched + exposes, columns and nested fields by name and kept the last of a repeated name, so an earlier + declaration was never compared. An expose id (`exposeId`, or `id`), a column name or a nested + field name that the new contract declares more than once is now `breaking` + (`expose_duplicate_id`, `column_duplicate_name`, `nested_field_duplicate_name`), in an expose both + contracts have and in one that is new, and whatever the baseline holds: a contract that already + declares a name twice is reported against itself until one declaration is removed, so a pipeline + that gates on it can start exiting 1, and the `fluid forge --refine` preview lists it as a + breaking change. + - **A failed `fluid apply` on OpenTofu printed raw JSON and no way forward.** The runner uses `-json`, so the error was the last 800 characters of diagnostic lines. It now reads the diagnostic (the summary, the resource address and the provider's detail) and says that OpenTofu diff --git a/fluid_build/api/changelog.py b/fluid_build/api/changelog.py index 767653ed5..8eefe8408 100644 --- a/fluid_build/api/changelog.py +++ b/fluid_build/api/changelog.py @@ -39,6 +39,7 @@ diff_agent_policy, diff_columns, diff_consumes, + diff_duplicates, diff_metadata, diff_quality_severity, diff_sovereignty, @@ -71,7 +72,9 @@ def compare_contracts( """Return a structured changelog comparing two parsed contracts. ``baseline`` is treated as "old"; ``new`` as the candidate. Idempotent — - passing the same dict twice produces an empty report. + passing the same dict twice produces an empty report — unless the contract + declares an expose id, a column name or a nested field name twice: that is + reported whatever the baseline holds (see :func:`diff_duplicates`). Both are first validated (:func:`_require_valid`), and an invalid one raises :class:`ContractValidationError`. ``validate=False`` diffs whatever @@ -98,6 +101,10 @@ def compare_contracts( report.extend(diff_agent_policy(baseline, new)) report.extend(diff_sovereignty(baseline, new)) + # A name the new contract declares twice: the comparison below keeps only + # the last declaration of it. + report.extend(diff_duplicates(new)) + # Per-expose schema and quality. for expose_id, idx, base_expose, new_expose in iter_expose_pairs(baseline, new): if base_expose is None and new_expose is not None: diff --git a/fluid_build/api/changelog_rules.py b/fluid_build/api/changelog_rules.py index a1733f33b..b4c8df43e 100644 --- a/fluid_build/api/changelog_rules.py +++ b/fluid_build/api/changelog_rules.py @@ -37,28 +37,82 @@ # types that strictly widen them. Anything not in this table is treated as # a breaking type change. Names normalized to upper-case AND stripped of # precision/scale annotations (those are compared separately below). +# +# Only conversions that keep every existing value are listed, and only into a +# target that declares no parameters of its own (see ``_classify_type_diff``). +# Apache Iceberg ("Schema Evolution" in the table spec) and Delta Lake ("Type +# widening") draw the same line: int -> long, float -> double and +# date -> timestamp (Iceberg v3, Delta timestampNTZ) widen; int -> float, +# float -> decimal and anything out of TIME are on neither list. _WIDENING: Mapping[str, frozenset[str]] = { - "INT": frozenset({"BIGINT", "INTEGER", "LONG", "DECIMAL", "NUMERIC", "FLOAT", "DOUBLE"}), - "INT32": frozenset({"INT64", "BIGINT", "LONG", "DECIMAL", "FLOAT", "DOUBLE"}), - "INTEGER": frozenset({"BIGINT", "LONG", "DECIMAL", "FLOAT", "DOUBLE"}), + "INT": frozenset({"BIGINT", "INTEGER", "LONG", "DECIMAL", "NUMERIC", "DOUBLE"}), + "INT32": frozenset({"INT64", "BIGINT", "LONG", "DECIMAL", "DOUBLE"}), + "INTEGER": frozenset({"BIGINT", "LONG", "DECIMAL", "DOUBLE"}), "SMALLINT": frozenset({"INT", "INT32", "INT64", "BIGINT", "INTEGER", "LONG"}), - "FLOAT": frozenset({"DOUBLE", "DECIMAL", "NUMERIC"}), + "FLOAT": frozenset({"DOUBLE"}), "VARCHAR": frozenset({"TEXT", "STRING"}), "STRING": frozenset({"TEXT"}), "DATE": frozenset({"DATETIME", "TIMESTAMP"}), - "TIME": frozenset({"DATETIME", "TIMESTAMP"}), } -# Type-name + parameter regex. Captures e.g. ``DECIMAL(10,2)`` as ("DECIMAL", -# "10,2") and ``VARCHAR(255)`` as ("VARCHAR", "255"). Suffix-attribute-less -# types ("BIGINT") fall through with empty params. -_TYPE_PARAM_RE = re.compile(r"^([A-Z0-9_]+)\s*(?:\(([^)]+)\))?\s*$") +# Fixed-point types: two parameters, (precision, scale). +_FIXED_POINT = frozenset({"DECIMAL", "DEC", "NUMERIC", "NUMBER", "BIGNUMERIC"}) + +# Types whose parameter measures capacity (a length or a precision), so a bigger +# one holds every value the smaller one did. The parameters of any other type +# name something instead (``GEOMETRY(4326)`` is a spatial reference id, +# ``VECTOR(1536)`` a dimension, ``BINARY(16)`` an exact width): changing them is +# not a widening, whichever way the number moves. +_MEASURED = _FIXED_POINT | frozenset( + { + "VARCHAR", + "VARCHAR2", + "NVARCHAR", + "NVARCHAR2", + "CHAR", + "NCHAR", + "CHARACTER", + "STRING", + "TEXT", + "VARBINARY", + "FLOAT", + "TIME", + "TIMESTAMP", + "TIMESTAMP_NTZ", + "TIMESTAMP_LTZ", + "TIMESTAMP_TZ", + "DATETIME", + "DATETIME2", + } +) + +# Type-name + parameter grammar. ASCII only, matched whole with ``fullmatch`` +# (``$`` would let a trailing newline through), and linear in the input: the one +# ``\s*`` sits inside the optional group, so a name followed by a long run of +# spaces can't backtrack quadratically (the input is stripped before it is +# matched). Captures e.g. ``DECIMAL(10,2)`` as ("DECIMAL", "10,2") and +# ``VARCHAR(255)`` as ("VARCHAR", "255"). Suffix-attribute-less types ("BIGINT") +# fall through with empty params. +_TYPE_PARAM_RE = re.compile(r"([A-Z0-9_]+)(?:\s*\(([^)]+)\))?", re.ASCII) + +# The parameters the classifier can order: plain ASCII integers. +_TYPE_INT_PARAMS_RE = re.compile(r"[0-9]+(?:\s*,\s*[0-9]+)*", re.ASCII) + +# The whitespace that may surround a parameter: ``str.strip()`` with no argument +# also removes non-ASCII spaces, which would let them pass for ASCII ones. +_ASCII_WS = " \t\n\r\f\v" def _norm_type(t: Any) -> str: if not isinstance(t, str): return "" - return t.strip().upper() + t = t.strip() + # Fold case for ASCII names only. ``str.upper()`` maps some non-ASCII + # letters onto ASCII ones (U+0131 dotless i -> "I", U+017F long s -> "S"), + # which would make an unknown type compare equal to a real one although the + # IaC emitters do not fold them. A non-ASCII name stays as written and + # matches nothing but itself. + return t.upper() if t.isascii() else t def _parse_type(t: str) -> Tuple[str, Optional[Tuple[int, ...]]]: @@ -70,21 +124,29 @@ def _parse_type(t: str) -> Tuple[str, Optional[Tuple[int, ...]]]: ``"BIGINT"`` → ``("BIGINT", None)`` ``"DECIMAL"`` → ``("DECIMAL", None)`` (no precision specified) - Returns ``("", None)`` when the input doesn't parse cleanly — caller - decides whether to treat that as a generic mismatch. + Returns ``("", None)`` for an empty input. A type that doesn't parse + cleanly (non-ASCII text, or parameters that aren't plain integers, such as + ``ENUM('A','B')`` or ``VARCHAR(MAX)``) comes back as ``(whole_string, + None)``: it equals itself and no other type, so a change in it is a + mismatch and never a normalization. """ if not t: return ("", None) - m = _TYPE_PARAM_RE.match(t.strip()) + t = t.strip() + m = _TYPE_PARAM_RE.fullmatch(t) if not m: - return (t.strip(), None) + return (t, None) base, params = m.group(1), m.group(2) if not params: return (base, None) + params = params.strip(_ASCII_WS) + if not _TYPE_INT_PARAMS_RE.fullmatch(params): + return (t, None) try: - nums = tuple(int(p.strip()) for p in params.split(",")) + nums = tuple(int(p) for p in params.split(",")) except ValueError: - return (base, None) + # Past the interpreter's integer-string limit. + return (t, None) return (base, nums) @@ -95,10 +157,15 @@ def _classify_param_change( ) -> Optional[str]: """Classify a same-base type parameter change as 'widening' / 'narrowing'. - Returns ``"widening"`` if every parameter grew (or stayed equal), - ``"narrowing"`` if any parameter shrank, ``"changed"`` for ambiguous / - not-comparable cases, or ``None`` when both sides have no params (so - there's no diff to classify). + Returns ``"widening"`` if no parameter shrank and the type's parameters + measure capacity (see ``_MEASURED``), ``"narrowing"`` if none grew, + ``"changed"`` for everything else (one side has no parameters, a different + number of them, parameters that moved in different directions, or ones that + name something rather than measure it), or ``None`` when both sides carry + the same parameters, or none (so there's no diff to classify). + + A fixed-point type's ``(precision, scale)`` is compared as (integer digits, + scale), not as the two numbers it is written with. """ if old_params is None and new_params is None: return None @@ -112,9 +179,19 @@ def _classify_param_change( if len(old_params) != len(new_params): # DECIMAL(10) vs DECIMAL(10,2) — semantically different shapes. return "changed" - pairs = list(zip(old_params, new_params, strict=True)) + if base in _FIXED_POINT and len(old_params) == 2: + # (precision, scale) holds ``precision - scale`` integer digits and + # ``scale`` fractional ones, so a bigger precision with a bigger scale + # can still shrink the integer part: (10,2) holds 99999999.99 and + # (11,4) holds 9999999.9999. Compare the two capacities. Delta Lake + # widens decimal(p,s) to (p+k1,s+k2) only when k1 >= k2 >= 0, the same + # condition; Iceberg widens precision at a fixed scale. + (o_prec, o_scale), (n_prec, n_scale) = old_params, new_params + pairs = [(o_prec - o_scale, n_prec - n_scale), (o_scale, n_scale)] + else: + pairs = list(zip(old_params, new_params, strict=True)) if all(n >= o for o, n in pairs): - return "widening" + return "widening" if base in _MEASURED else "changed" if all(n <= o for o, n in pairs): return "narrowing" # Mixed: e.g. precision grew but scale shrank — neither pure widening @@ -134,16 +211,26 @@ def _is_widening(old_type: str, new_type: str) -> bool: return new_type in widens_to +def _why_not_widening(old_type: str, new_type: str) -> str: + """The reason, for a finding's text, a type change is not a widening.""" + if old_type.isascii() and new_type.isascii(): + return "not a widening cast" + return "a type name with non-ASCII characters matches no other type" + + def _classify_type_diff(old_full: str, new_full: str) -> Tuple[str, str]: """Classify a (possibly parameterized) type change. Returns ``(severity, kind)``: - * ``("non_breaking", "column_type_widened")`` — base widened or - parameters strictly grew + * ``("non_breaking", "column_type_widened")`` — base widened to a type + that declares no parameters, or parameters strictly grew * ``("breaking", "column_type_changed")`` — base narrowed, parameters shrank, or unrelated type swap - * ``("info", "column_type_param_changed")`` — same base, parameters - changed in a way that's neither pure widening nor pure narrowing + * ``("breaking", "column_type_param_changed")`` — same base, parameters + changed in a way that's neither pure widening nor pure narrowing, or + only one side declares parameters + * ``("info", "column_type_normalized")`` — same base and parameters, + written with different formatting (whitespace, leading zeros) """ old_base, old_params = _parse_type(old_full) new_base, new_params = _parse_type(new_full) @@ -153,30 +240,85 @@ def _classify_type_diff(old_full: str, new_full: str) -> Tuple[str, str]: change = _classify_param_change(old_params, new_params, old_base) if change is None: # Same base, no param difference but full strings differ — - # whitespace or capitalization. Treat as info-level. + # whitespace or leading zeros (case is folded before this point). return ("info", "column_type_normalized") if change == "widening": return ("non_breaking", "column_type_widened") if change == "narrowing": return ("breaking", "column_type_changed") - return ("info", "column_type_param_changed") + return ("breaking", "column_type_param_changed") - # Different base type — fall back to the base-name widening table. - if _is_widening(old_base, new_base): + # Different base type — fall back to the base-name widening table. The + # table knows base names only: a target that declares parameters + # (``INT`` -> ``DECIMAL(5,0)``) may hold less than the source did. + if new_params is None and _is_widening(old_base, new_base): return ("non_breaking", "column_type_widened") return ("breaking", "column_type_changed") +def _column_name(col: Any) -> Optional[str]: + """The name a column is matched by: ``name``, else ``id``.""" + if isinstance(col, Mapping): + name = col.get("name") or col.get("id") + if isinstance(name, str): + return name + return None + + def _columns_by_name(columns: Sequence[Mapping[str, Any]]) -> dict[str, Mapping[str, Any]]: + """Index columns by name; the last of a repeated name wins. + + :func:`diff_duplicates` reports the names this would collapse. + """ out: dict[str, Mapping[str, Any]] = {} for col in columns or []: - if isinstance(col, Mapping): - name = col.get("name") or col.get("id") - if isinstance(name, str): - out[name] = col + name = _column_name(col) + if name is not None: + out[name] = col return out +def _duplicate_columns(columns: Sequence[Mapping[str, Any]]) -> dict[str, list[Mapping[str, Any]]]: + """Every name declared more than once, with all of its declarations.""" + seen: dict[str, list[Mapping[str, Any]]] = {} + if not isinstance(columns, Sequence): + return {} + for col in columns: + name = _column_name(col) + if name is not None: + seen.setdefault(name, []).append(col) + return {name: cols for name, cols in seen.items() if len(cols) > 1} + + +def _duplicate_changes( + declarations: Sequence[Mapping[str, Any]], + *, + kind: str, + noun: str, + expose_id: str, + path_prefix: str, +) -> List[Change]: + """A breaking ``Change`` for every name the new contract declares more than once. + + :func:`_columns_by_name` keeps the last declaration of a name, so without this + the earlier one would never be compared against the baseline. + """ + return [ + Change( + path=f"{path_prefix}.{name}", + kind=kind, + severity="breaking", + description=( + f"{noun} '{name}' is declared {len(declared)} times in expose " + f"'{expose_id}'; keep one (a later declaration hides an earlier one)" + ), + before=None, + after=[dict(d) for d in declared], + ) + for name, declared in _duplicate_columns(declarations).items() + ] + + def _expose_schema(expose: Mapping[str, Any]) -> Tuple[Sequence[Any], Optional[str]]: """Return an expose's columns and where, below the expose, they were found. @@ -417,17 +559,17 @@ def diff_columns( elif kind == "column_type_param_changed": desc = ( f"column '{col_name}' type parameters changed " - f"{old_type} -> {new_type} (review: not pure widening or narrowing)" + f"{old_type} -> {new_type} (not a pure widening)" ) elif kind == "column_type_normalized": desc = ( f"column '{col_name}' type normalized " - f"{old_type} -> {new_type} (whitespace/casing only)" + f"{old_type} -> {new_type} (formatting only)" ) else: desc = ( f"column '{col_name}' type changed " - f"{old_type} -> {new_type} (not a widening cast)" + f"{old_type} -> {new_type} ({_why_not_widening(old_type, new_type)})" ) changes.append( Change( @@ -948,14 +1090,10 @@ def iter_expose_pairs( base_exposes = list(baseline.get("exposes") or []) new_exposes = list(new.get("exposes") or []) base_by_id = { - (e.get("id") or e.get("exposeId") or f"_pos_{i}"): (i, e) - for i, e in enumerate(base_exposes) - if isinstance(e, Mapping) + _expose_key(i, e): (i, e) for i, e in enumerate(base_exposes) if isinstance(e, Mapping) } new_by_id = { - (e.get("id") or e.get("exposeId") or f"_pos_{i}"): (i, e) - for i, e in enumerate(new_exposes) - if isinstance(e, Mapping) + _expose_key(i, e): (i, e) for i, e in enumerate(new_exposes) if isinstance(e, Mapping) } all_ids = set(base_by_id) | set(new_by_id) for eid in all_ids: @@ -963,3 +1101,89 @@ def iter_expose_pairs( n = new_by_id.get(eid) idx_in_new = n[0] if n else (b[0] if b else 0) yield eid, idx_in_new, (b[1] if b else None), (n[1] if n else None) + + +def _expose_key(index: int, expose: Mapping[str, Any]) -> Any: + """The id exposes are matched by; one without an id is matched by position.""" + return expose.get("id") or expose.get("exposeId") or f"_pos_{index}" + + +def _nested_duplicates(columns: Sequence[Any], *, expose_id: str, path_prefix: str) -> List[Change]: + """Every ``fields`` name declared twice below *columns*, at any depth.""" + changes: List[Change] = [] + if not isinstance(columns, Sequence): + return changes + for col in columns: + name = _column_name(col) + fields = col.get("fields") if name is not None else None + if isinstance(fields, Sequence): + prefix = f"{path_prefix}.{name}.fields" + changes.extend( + _duplicate_changes( + fields, + kind="nested_field_duplicate_name", + noun="nested field", + expose_id=expose_id, + path_prefix=prefix, + ) + ) + changes.extend(_nested_duplicates(fields, expose_id=expose_id, path_prefix=prefix)) + return changes + + +def _duplicate_exposes(exposes: Sequence[Any]) -> List[Change]: + """Every expose id declared twice. An expose without an id is matched by position.""" + declared_at: dict[Any, list[int]] = {} + for i, expose in enumerate(exposes): + if isinstance(expose, Mapping): + expose_id = expose.get("id") or expose.get("exposeId") + if expose_id: + declared_at.setdefault(expose_id, []).append(i) + return [ + Change( + path=f"exposes[{at[-1]}]", + kind="expose_duplicate_id", + severity="breaking", + description=( + f"expose '{expose_id}' is declared {len(at)} times " + f"(at {', '.join(f'exposes[{i}]' for i in at)}); " + f"keep one (a later declaration hides an earlier one)" + ), + before=None, + after={"id": expose_id, "indexes": at}, + ) + for expose_id, at in declared_at.items() + if len(at) > 1 + ] + + +def diff_duplicates(new: Mapping[str, Any]) -> List[Change]: + """Flag every expose id, column name and nested field name *new* declares twice. + + Breaking, and reported whatever the baseline holds. The comparison indexes + exposes, columns and fields by name and keeps the last of a repeated name, + so an earlier declaration would never be compared against the baseline, + and an emitter is free to read either one. It follows that a contract that + already declares a name twice is reported against itself, until one + declaration is removed. + """ + exposes = list(new.get("exposes") or []) + changes = _duplicate_exposes(exposes) + for i, expose in enumerate(exposes): + if not isinstance(expose, Mapping): + continue + columns, where_found = _expose_schema(expose) + if where_found is None: + continue + key, prefix = _expose_key(i, expose), f"exposes[{i}].{where_found}" + changes.extend( + _duplicate_changes( + columns, + kind="column_duplicate_name", + noun="column", + expose_id=key, + path_prefix=prefix, + ) + ) + changes.extend(_nested_duplicates(columns, expose_id=key, path_prefix=prefix)) + return changes diff --git a/tests/test_diff_changelog.py b/tests/test_diff_changelog.py index f4e03f6cc..53a754c3f 100644 --- a/tests/test_diff_changelog.py +++ b/tests/test_diff_changelog.py @@ -21,8 +21,11 @@ from __future__ import annotations import argparse +import copy import json import logging +import sys +import time from pathlib import Path import pytest @@ -30,10 +33,12 @@ from fluid_build.api.changelog import compare_contracts, render_markdown, render_text from fluid_build.api.changelog_rules import ( _classify_type_diff, + _norm_type, _parse_type, diff_agent_policy, diff_columns, diff_consumes, + diff_duplicates, diff_quality_severity, diff_sovereignty, ) @@ -42,6 +47,14 @@ FIXTURES = Path(__file__).parent / "fixtures" / "changelog" +# Characters the type tests need that look like ASCII, built with chr() so the source +# stays plain ASCII and greppable. +DOTLESS_I = chr(0x131) # str.upper() maps it to "I" +LONG_S = chr(0x17F) # str.upper() maps it to "S" +DOTTED_CAPITAL_I = chr(0x130) +FULLWIDTH_ONE = chr(0xFF11) +NBSP = chr(0xA0) + @pytest.fixture def v1_v2(): @@ -475,6 +488,14 @@ def test_cli_diff_baseline_rejects_state(tmp_path): ("BIGINT", "BIGINT", None), ("DECIMAL(38, 18)", "DECIMAL", (38, 18)), ("", "", None), + # Off-grammar text is its own type: it equals itself and nothing else, so a + # change in it can never be read as "no parameters" or as whitespace. + ("ENUM('A','B')", "ENUM('A','B')", None), + ("VARCHAR(MAX)", "VARCHAR(MAX)", None), + ("DECIMAL(1_0,2)", "DECIMAL(1_0,2)", None), + ("DECIMAL(+10,2)", "DECIMAL(+10,2)", None), + (f"DECIMAL({FULLWIDTH_ONE}0,2)", f"DECIMAL({FULLWIDTH_ONE}0,2)", None), + (f"DECIMAL({NBSP}10,2)", f"DECIMAL({NBSP}10,2)", None), ], ) def test_parse_type_extracts_base_and_params(raw, expected_base, expected_params): @@ -495,10 +516,14 @@ def test_decimal_precision_narrowing_is_breaking(): assert kind == "column_type_changed" -def test_decimal_mixed_param_change_is_info(): - """Precision grew but scale shrank — neither pure widening nor narrowing.""" +def test_decimal_mixed_param_change_is_breaking(): + """Precision grew but scale shrank — neither pure widening nor narrowing. + + It used to be ``info``, which ``--fail-on-breaking`` lets through; a column that + holds fewer fractional digits than before loses data for its consumers. + """ sev, kind = _classify_type_diff("DECIMAL(20,4)", "DECIMAL(30,2)") - assert sev == "info" + assert sev == "breaking" assert kind == "column_type_param_changed" @@ -894,24 +919,18 @@ def test_a_column_dropped_while_moving_to_contract_schema_is_still_found(): ] -@pytest.mark.parametrize("fail_on_breaking, expected_rc", [(True, 1), (False, 0)]) -def test_cli_diff_baseline_sees_a_required_column_added_to_a_076_contract( - tmp_path, fail_on_breaking, expected_rc -): - """``--fail-on-breaking`` is the exit code CI gates on; it must see 0.7.x columns.""" +def _cli_diff_baseline(tmp_path, old, new, fail_on_breaking): + """Run ``fluid diff --baseline `` on two contract dicts; return (rc, JSON report).""" import yaml from fluid_build.cli import diff as diff_mod - old = tmp_path / "old.fluid.yaml" - new = tmp_path / "new.fluid.yaml" - old.write_text(yaml.safe_dump(_contract_076([_ID_COLUMN]))) - added = {"name": "region", "type": "string", "required": True} - new.write_text(yaml.safe_dump(_contract_076([_ID_COLUMN, added]))) + (tmp_path / "old.fluid.yaml").write_text(yaml.safe_dump(old)) + (tmp_path / "new.fluid.yaml").write_text(yaml.safe_dump(new)) out = tmp_path / "diff.json" args = argparse.Namespace( - contract=str(new), - baseline=str(old), + contract=str(tmp_path / "new.fluid.yaml"), + baseline=str(tmp_path / "old.fluid.yaml"), env=None, state=None, out=str(out), @@ -920,11 +939,386 @@ def test_cli_diff_baseline_sees_a_required_column_added_to_a_076_contract( format="json", cmd="diff", ) - rc = diff_mod.run(args, logging.getLogger("test.diff")) + return rc, json.loads(out.read_text()) + + +@pytest.mark.parametrize("fail_on_breaking, expected_rc", [(True, 1), (False, 0)]) +def test_cli_diff_baseline_sees_a_required_column_added_to_a_076_contract( + tmp_path, fail_on_breaking, expected_rc +): + """``--fail-on-breaking`` is the exit code CI gates on; it must see 0.7.x columns.""" + added = {"name": "region", "type": "string", "required": True} + + rc, report = _cli_diff_baseline( + tmp_path, _contract_076([_ID_COLUMN]), _contract_076([_ID_COLUMN, added]), fail_on_breaking + ) + + assert rc == expected_rc + assert [c["kind"] for c in report["breaking"]] == ["column_added"] + + +# --------------------------------------------------------------------------- +# Type comparison: type changes that used to pass ``--fail-on-breaking`` +# --------------------------------------------------------------------------- + + +def _assert_schema_valid(*contracts): + """Every contract below is one the bundled 0.7.6 schema accepts, so the engine must handle it.""" + from fluid_build.schema_manager import FluidSchemaManager + + for contract in contracts: + result = FluidSchemaManager().validate_contract( + contract, schema_version="0.7.6", offline_only=True + ) + assert result.is_valid, result.errors + + +def _type_change_report(old_type, new_type): + old = _contract_076([{"name": "c", "type": old_type}]) + new = _contract_076([{"name": "c", "type": new_type}]) + _assert_schema_valid(old, new) + return compare_contracts(old, new) + + +@pytest.mark.parametrize( + "old_type, new_type, severity, kind", + [ + # A name that only looks like a real type once str.upper() has folded it. + pytest.param("int", f"{DOTLESS_I}nt", "breaking", "column_type_changed", id="dotless-i"), + pytest.param( + "timestamp", f"time{LONG_S}tamp", "breaking", "column_type_changed", id="long-s" + ), + # Parameters used to be ignored when the base type changed. + ("int", "decimal(5,0)", "breaking", "column_type_changed"), + # Parameters present on one side only, or moving in different directions. + ("varchar", "varchar(10)", "breaking", "column_type_param_changed"), + ("varchar(255)", "varchar", "breaking", "column_type_param_changed"), + ("decimal(10,2)", "decimal(12,1)", "breaking", "column_type_param_changed"), + ("decimal(10,2)", "decimal", "breaking", "column_type_param_changed"), + # Precision and scale both grow, the integer digits shrink (8 -> 7). + ("decimal(10,2)", "decimal(11,4)", "breaking", "column_type_param_changed"), + # Parameters that name something (a spatial reference id, an exact width) don't widen. + ("geometry(4326)", "geometry(32633)", "breaking", "column_type_param_changed"), + ("binary(16)", "binary(32)", "breaking", "column_type_param_changed"), + # Entries that left the widening table. + ("time", "timestamp", "breaking", "column_type_changed"), + ("time", "datetime", "breaking", "column_type_changed"), + ("int", "float", "breaking", "column_type_changed"), + ("integer", "float", "breaking", "column_type_changed"), + ("int32", "float", "breaking", "column_type_changed"), + ("float", "decimal", "breaking", "column_type_changed"), + ("float", "numeric", "breaking", "column_type_changed"), + # Parameters the classifier can't order used to read as "no parameters". + ("enum('a','b')", "enum('a')", "breaking", "column_type_changed"), + ("varchar", "varchar(max)", "breaking", "column_type_changed"), + ("decimal(10,2)", "decimal(1_0,2)", "breaking", "column_type_changed"), + ("decimal(10,2)", f"decimal({NBSP}10,2)", "breaking", "column_type_changed"), + # Controls, which pass on the code before this change too: what still widens, and + # what is only whitespace. + ("int", "bigint", "non_breaking", "column_type_widened"), + ("int", "double", "non_breaking", "column_type_widened"), + ("int", "decimal", "non_breaking", "column_type_widened"), + ("varchar", "text", "non_breaking", "column_type_widened"), + ("varchar(50)", "varchar(255)", "non_breaking", "column_type_widened"), + ("date", "timestamp", "non_breaking", "column_type_widened"), + ("float", "double", "non_breaking", "column_type_widened"), + ("decimal(10,2)", "decimal(12,4)", "non_breaking", "column_type_widened"), + ("char(10)", "char(20)", "non_breaking", "column_type_widened"), + ("float(24)", "float(53)", "non_breaking", "column_type_widened"), + ("decimal(10,2)", "decimal(10, 2)", "info", "column_type_normalized"), + ("varchar(255)", "varchar(0255)", "info", "column_type_normalized"), + ], +) +def test_076_type_comparison(old_type, new_type, severity, kind): + report = _type_change_report(old_type, new_type) + + assert report.total == 1 + (change,) = getattr(report, severity) + assert change.kind == kind + assert change.path == "exposes[0].contract.schema.c.type" + + +@pytest.mark.parametrize( + "raw, expected", + [ + (" int ", "INT"), + ("DeCiMaL(10,2)", "DECIMAL(10,2)"), + (f"{DOTLESS_I}nt", f"{DOTLESS_I}nt"), + (f"time{LONG_S}tamp", f"time{LONG_S}tamp"), + (f"{DOTTED_CAPITAL_I}nt", f"{DOTTED_CAPITAL_I}nt"), + (None, ""), + (5, ""), + ], +) +def test_norm_type_folds_case_for_ascii_names_only(raw, expected): + assert _norm_type(raw) == expected + + +@pytest.mark.parametrize( + "old, new, expected", + [ + # Same integer digits, bigger scale: still holds every old value. + ("DECIMAL(10,2)", "DECIMAL(12,4)", ("non_breaking", "column_type_widened")), + ("DECIMAL(10,2)", "DECIMAL(11,4)", ("breaking", "column_type_param_changed")), + ("NUMBER(10,2)", "NUMBER(11,4)", ("breaking", "column_type_param_changed")), + ("NUMERIC(10,2)", "NUMERIC(12,1)", ("breaking", "column_type_param_changed")), + ("DECIMAL(10)", "DECIMAL(10,2)", ("breaking", "column_type_param_changed")), + ("DECIMAL(10)", "DECIMAL(12)", ("non_breaking", "column_type_widened")), + # Types that are not fixed-point compare their parameters one by one. + ("TIMESTAMP(3)", "TIMESTAMP(6)", ("non_breaking", "column_type_widened")), + ], +) +def test_fixed_point_widens_only_when_scale_and_integer_digits_both_hold(old, new, expected): + assert _classify_type_diff(old, new) == expected + + +@pytest.mark.parametrize( + "old, new", + [ + ("GEOMETRY(4326)", "GEOMETRY(32633)"), + ("GEOMETRY(32633)", "GEOMETRY(4326)"), + ("VECTOR(3)", "VECTOR(1536)"), + ("BINARY(16)", "BINARY(32)"), + ("BIT(8)", "BIT(16)"), + ], +) +def test_parameters_that_name_something_are_never_a_widening(old, new): + """A spatial reference id, a dimension or an exact width does not measure capacity.""" + assert _classify_type_diff(old, new)[0] == "breaking" + + +@pytest.mark.parametrize( + "old, new", + [ + ("VARCHAR(10)", "VARCHAR(20)"), + ("CHAR(1)", "CHAR(8)"), + ("FLOAT(24)", "FLOAT(53)"), + ("TIME(0)", "TIME(6)"), + ("DATETIME2(3)", "DATETIME2(7)"), + ], +) +def test_a_bigger_length_or_precision_is_a_widening(old, new): + assert _classify_type_diff(old, new) == ("non_breaking", "column_type_widened") + + +def test_parsing_a_long_run_of_spaces_or_an_unclosed_list_stays_linear(): + """A type is untrusted text: the parser must not backtrack quadratically on it.""" + started = time.perf_counter() + _parse_type("A" + " " * 50_000 + "x") + _parse_type("A(" + "1" * 50_000) + assert time.perf_counter() - started < 2.0 + + +def test_an_oversized_integer_parameter_is_unparseable_text(): + if not hasattr(sys, "set_int_max_str_digits"): + pytest.skip("this interpreter has no integer-string length limit") + previous = sys.get_int_max_str_digits() + sys.set_int_max_str_digits(640) + try: + raw = "VARCHAR(" + "9" * 700 + ")" + assert _parse_type(raw) == (raw, None) + finally: + sys.set_int_max_str_digits(previous) + + +def test_a_non_ascii_type_name_says_why_it_is_not_the_same_type(): + report = _type_change_report("int", f"{DOTLESS_I}nt") + + (change,) = report.breaking + assert "non-ASCII" in change.description + + +@pytest.mark.parametrize("fail_on_breaking, expected_rc", [(True, 1), (False, 0)]) +def test_cli_diff_baseline_fails_on_a_type_change_that_used_to_pass( + tmp_path, fail_on_breaking, expected_rc +): + """``int`` -> ``decimal(5,0)`` was reported as a widening, so the CI gate stayed green.""" + old = _contract_076([{"name": "id", "type": "int"}]) + new = _contract_076([{"name": "id", "type": "decimal(5,0)"}]) + _assert_schema_valid(old, new) + + rc, report = _cli_diff_baseline(tmp_path, old, new, fail_on_breaking) assert rc == expected_rc - assert [c["kind"] for c in json.loads(out.read_text())["breaking"]] == ["column_added"] + assert [c["kind"] for c in report["breaking"]] == ["column_type_changed"] + + +def _contract_with_exposes(*exposes): + """A 0.7.6 contract with one expose per ``(expose_id, columns)`` pair.""" + contract = _contract_076([]) + template = contract["exposes"][0] + contract["exposes"] = [] + for expose_id, columns in exposes: + expose = copy.deepcopy(template) + expose["exposeId"] = expose_id + expose["contract"]["schema"] = columns + contract["exposes"].append(expose) + return contract + + +@pytest.mark.parametrize( + "declared, kinds", + [ + pytest.param(["int", "string"], {"column_duplicate_name"}, id="last-matches-baseline"), + pytest.param( + ["string", "int"], {"column_duplicate_name", "column_type_changed"}, id="last-differs" + ), + pytest.param(["string", "string"], {"column_duplicate_name"}, id="identical"), + pytest.param(["string"] * 3, {"column_duplicate_name"}, id="three-times"), + ], +) +def test_076_a_column_declared_twice_in_the_new_contract_is_breaking(declared, kinds): + """The index keeps the last declaration of a name, so the diff never saw the other one.""" + old = _contract_076([_ID_COLUMN, {"name": "note", "type": "string"}]) + new = _contract_076([_ID_COLUMN] + [{"name": "note", "type": t} for t in declared]) + _assert_schema_valid(old, new) + + report = compare_contracts(old, new) + + assert {c.kind for c in report.breaking} == kinds + assert report.total == len(kinds) + (duplicate,) = [c for c in report.breaking if c.kind == "column_duplicate_name"] + assert duplicate.path == "exposes[0].contract.schema.note" + assert [c["type"] for c in duplicate.after] == declared + + +_ZIP = {"name": "zip", "type": "STRING"} + + +def _address(*fields): + return {"name": "addr", "type": "STRUCT", "fields": list(fields)} + + +@pytest.mark.parametrize( + "old_columns, new_columns, non_breaking", + [ + pytest.param( + [_address(_ZIP)], + [_address({"name": "zip", "type": "INT"}, _ZIP)], + [], + id="column-in-both-contracts", + ), + pytest.param( + [{"name": "id", "type": "STRING"}], + [{"name": "id", "type": "STRING"}, _address(_ZIP, _ZIP)], + ["column_added"], + id="column-is-new", + ), + ], +) +def test_a_nested_field_declared_twice_in_the_new_contract_is_breaking( + old_columns, new_columns, non_breaking +): + report = compare_contracts( + _contract_flat(old_columns), _contract_flat(new_columns), validate=False + ) + + assert [c.kind for c in report.non_breaking] == non_breaking + assert [(c.kind, c.path) for c in report.breaking] == [ + ("nested_field_duplicate_name", "exposes[0].schema.addr.fields.zip") + ] + + +def test_076_an_expose_id_declared_twice_in_the_new_contract_is_breaking(): + as_string = [{"name": "a", "type": "string"}] + as_int = [{"name": "a", "type": "int"}] + old = _contract_with_exposes(("orders", as_string)) + new = _contract_with_exposes(("orders", as_int), ("orders", as_string)) + _assert_schema_valid(old, new) + + report = compare_contracts(old, new) + + assert report.total == 1 + (change,) = report.breaking + assert (change.kind, change.path) == ("expose_duplicate_id", "exposes[1]") + assert change.after == {"id": "orders", "indexes": [0, 1]} + + +def test_076_distinct_names_and_ids_are_not_duplicates(): + columns = [_ID_COLUMN, {"name": "note", "type": "string"}] + contract = _contract_with_exposes(("orders", columns), ("customers", columns)) + _assert_schema_valid(contract) + + assert compare_contracts(contract, copy.deepcopy(contract)).total == 0 + + +def test_a_duplicate_in_an_expose_that_is_new_to_the_contract_is_breaking(): + """A new expose has no baseline to hide a change from, but it is just as ambiguous.""" + customers = ("customers", [_ID_COLUMN]) + note = {"name": "note", "type": "string"} + old = _contract_with_exposes(customers) + new = _contract_with_exposes(customers, ("orders", [_ID_COLUMN, note, note])) + _assert_schema_valid(old, new) + + report = compare_contracts(old, new) + + assert [c.kind for c in report.non_breaking] == ["expose_added"] + assert [(c.kind, c.path) for c in report.breaking] == [ + ("column_duplicate_name", "exposes[1].contract.schema.note") + ] + + +def test_a_nested_field_declared_twice_below_the_first_level_is_breaking(): + lat = {"name": "lat", "type": "DOUBLE"} + geo = {"name": "geo", "type": "STRUCT", "fields": [lat, lat]} + contract = _contract_flat([{"name": "addr", "type": "STRUCT", "fields": [geo]}]) + + assert [(c.kind, c.path) for c in diff_duplicates(contract)] == [ + ("nested_field_duplicate_name", "exposes[0].schema.addr.fields.geo.fields.lat") + ] + + +def test_076_a_duplicate_both_contracts_declare_is_still_reported(): + """The documented exception to idempotence: the earlier declaration is hidden in both.""" + note = {"name": "note", "type": "string"} + contract = _contract_076([_ID_COLUMN, note, note]) + _assert_schema_valid(contract) + + report = compare_contracts(contract, copy.deepcopy(contract)) + + assert [c.kind for c in report.breaking] == ["column_duplicate_name"] + assert report.total == 1 + + +def test_076_a_duplicate_only_the_baseline_declares_is_not_reported(): + note = {"name": "note", "type": "string"} + old = _contract_076([_ID_COLUMN, note, note]) + new = _contract_076([_ID_COLUMN, note]) + + assert compare_contracts(old, new).total == 0 + + +def test_an_expose_is_named_by_id_or_exposeid_and_an_unnamed_one_by_position(): + exposes = [{"id": "a"}, {"exposeId": "a"}, {"id": "b"}, {"schema": []}, {"id": "_pos_3"}] + + found = diff_duplicates({"exposes": exposes}) + + assert [(c.kind, c.path) for c in found] == [("expose_duplicate_id", "exposes[1]")] + + +@pytest.mark.parametrize("schema", [5, True, 3.5, "text", {"x": 1}]) +def test_a_malformed_schema_in_an_expose_that_is_new_does_not_crash(schema): + """The duplicate walk reads exposes the comparison used to skip as merely added. + + A schema-invalid contract only gets here with ``validate=False``. + """ + old = {"exposes": [{"id": "a", "schema": [{"name": "x", "type": "INT"}]}]} + new = {"exposes": [old["exposes"][0], {"id": "b", "schema": schema}]} + + report = compare_contracts(old, new, validate=False) + + assert [c.kind for c in report.non_breaking] == ["expose_added"] + assert report.total == 1 + + +def test_the_duplicate_findings_render(): + note = {"name": "note", "type": "string"} + report = compare_contracts(_contract_076([_ID_COLUMN]), _contract_076([_ID_COLUMN, note, note])) + + assert "column_duplicate_name" in render_text(report) + assert "column_duplicate_name" in render_markdown(report) + assert json.loads(json.dumps(report.to_dict()))["breaking"][0]["after"] == [note, note] # --------------------------------------------------------------------------- From 3fa5f4c73411dbf501eb1e0f790d5b78d5144cb7 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 4 Nov 2026 06:18:10 +0100 Subject: [PATCH 5/6] fix(streaming): build the Flink config validator with schema_validator #811 added a guard, test_every_validator_is_built_by_schema_validator, which fails on any validator built outside schema_manager.schema_validator. streaming/compiler.py built a plain Draft202012Validator for parameters.flink, so the merged tree failed that guard. Building it through schema_validator keeps Draft 2020-12 as the dialect and makes each pattern end at the end of the value. So a trailing newline in deployment.name or image is now refused; #809's allowlists did not cover those two fields. Two test_fail_closed cases pin it (red before this change). --- fluid_build/streaming/compiler.py | 8 +++++--- tests/streaming/test_compiler.py | 9 +++++++++ 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/fluid_build/streaming/compiler.py b/fluid_build/streaming/compiler.py index 4ea41bc1f..aa770eea2 100644 --- a/fluid_build/streaming/compiler.py +++ b/fluid_build/streaming/compiler.py @@ -35,6 +35,7 @@ ) from fluid_build.providers._iceberg_catalog_auth import FORBIDDEN_TEXT_PATTERN, parse_catalog_auth from fluid_build.providers._sql_safety import quote_ansi_string_literal, validate_ident +from fluid_build.schema_manager import schema_validator from fluid_build.streaming.types import LogicalType, columns_sql, parse_fields from fluid_build.util.contract import is_externally_run_build @@ -224,9 +225,10 @@ def compile_build(contract: Mapping[str, Any], build: Mapping[str, Any]) -> Flin cfg = managed_config(build) if cfg is None: raise ValueError("This Flink build is externally managed; opt in with parameters.flink") - errors = sorted( - jsonschema.Draft202012Validator(_CONFIG).iter_errors(cfg), key=lambda e: str(e.path) - ) + # schema_validator, so that a pattern ends at the end of the value, not + # before a trailing newline as jsonschema's own re.search lets it. + validator = schema_validator(_CONFIG, default=jsonschema.Draft202012Validator) + errors = sorted(validator.iter_errors(cfg), key=lambda e: str(e.path)) if errors: # Do not echo instance values: an invalid value might be a pasted credential. problem = errors[0] diff --git a/tests/streaming/test_compiler.py b/tests/streaming/test_compiler.py index cf234e0cb..189804e05 100644 --- a/tests/streaming/test_compiler.py +++ b/tests/streaming/test_compiler.py @@ -57,6 +57,15 @@ def test_sql_and_runtime_changes_change_digest(contract): "mutate,match", [ (lambda c, f: f["deployment"].update(image="flink:latest"), "invalid configuration"), + # A pattern ends at the end of the value: a trailing newline is refused. + ( + lambda c, f: f["deployment"].update(name=f["deployment"]["name"] + "\n"), + "invalid configuration", + ), + ( + lambda c, f: f["deployment"].update(image=f["deployment"]["image"] + "\n"), + "invalid configuration", + ), (lambda c, f: f["deployment"].pop("context"), "invalid configuration"), (lambda c, f: f.update(typo=True), "invalid configuration"), (lambda c, f: f["source"].update(format="protobuf"), "invalid configuration"), From 85174c04d9f4ebeb6a8a151e8322f499afc1cc74 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Wed, 4 Nov 2026 06:35:29 +0100 Subject: [PATCH 6/6] security(credentials): litellm 1.108.0's seven new provider endpoints are dotenv-protected (#820) litellm 1.108.0 (released 2026-11-04) reads BESPOKE_API_BASE, LAYA_API_BASE, DATABRICKS_AGENT_API_BASE and MICROSOFT_365_COPILOT_API_BASE (new litellm.provider_list entries), CORALBRICKS_API_BASE and INFERENCEAPIS_API_BASE (api_base_env in llms/openai_like/providers.json), and SCALEDOWN_API_BASE (get_secret_str in llms/scaledown/chat/ transformation.py, "the host in SCALEDOWN_API_BASE"). Each picks the host the provider key goes to, so each joins LITELLM_ENDPOINT_NAMES. tests/test_litellm_endpoint_names.py failed on every CI leg that installs the newest litellm (main and all open PRs); it passes against the 1.108.0 wheel (58 passed, 1 failed without the change). --- CHANGELOG.md | 7 +++++++ fluid_build/credentials/litellm_endpoints.py | 13 ++++++++++--- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index aa0a66041..b0c463527 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -614,6 +614,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 '93.184.216.34', is public`. A URL that cannot be parsed, such as `http://[::1`, is refused with "it has no host" where it raised `ValueError`. +- **litellm 1.108.0 reads seven provider endpoints the dotenv guard did not know.** + `BESPOKE_API_BASE`, `CORALBRICKS_API_BASE`, `DATABRICKS_AGENT_API_BASE`, `INFERENCEAPIS_API_BASE`, + `LAYA_API_BASE`, `MICROSOFT_365_COPILOT_API_BASE` and `SCALEDOWN_API_BASE` choose the host + litellm sends the provider key to. They are now in `credentials/litellm_endpoints.py`, so a + project dotenv file cannot point them elsewhere while `FLUID_LLM_API_KEY` or a stored key is in + play, as for every other provider endpoint. + ### Upgrade notes - New validate warning, so `--strict` fails: a contract with builds whose `contract.id` or a diff --git a/fluid_build/credentials/litellm_endpoints.py b/fluid_build/credentials/litellm_endpoints.py index 67af1e4c8..53d71d680 100644 --- a/fluid_build/credentials/litellm_endpoints.py +++ b/fluid_build/credentials/litellm_endpoints.py @@ -24,9 +24,9 @@ project's own setting and follows the name-shape rule. Vendored because importing litellm takes seconds and dotenv hydration runs on -every command. Generated from the litellm 1.83.14, 1.97.0, 1.105.0 and -1.106.0 wheels (FLUID's floor, the last release Python 3.10 can import, and the -two newest), as the union of: +every command. Generated from the litellm 1.83.14, 1.97.0, 1.105.0, 1.106.0 +and 1.108.0 wheels (FLUID's floor, the last release Python 3.10 can import, and +the newest), as the union of: * ``_API_BASE`` for every ``litellm.provider_list`` entry (``-`` read as ``_``); @@ -83,6 +83,7 @@ "BASETEN_API_BASE", "BEDROCK_API_BASE", "BEDROCK_MANTLE_API_BASE", + "BESPOKE_API_BASE", "BFL_API_BASE", "BLACK_FOREST_LABS_API_BASE", "BRAVE_API_BASE", @@ -100,6 +101,7 @@ "COMETAPI_API_BASE", "COMETAPI_BASE_URL", "COMPACTIFAI_API_BASE", + "CORALBRICKS_API_BASE", "CORTECS_API_BASE", "CRUSOE_API_BASE", "CRW_API_BASE", @@ -108,6 +110,7 @@ "CUSTOM_OPENAI_API_BASE", "DARKBLOOM_API_BASE", "DASHSCOPE_API_BASE", + "DATABRICKS_AGENT_API_BASE", "DATABRICKS_API_BASE", "DATAFORSEO_API_BASE", "DATAROBOT_API_BASE", @@ -156,6 +159,7 @@ "HUMANLOOP_API_BASE", "HYPERBOLIC_API_BASE", "INCEPTION_API_BASE", + "INFERENCEAPIS_API_BASE", "INFINITY_API_BASE", "JINA_AI_API_BASE", "LAMBDA_AI_API_BASE", @@ -163,6 +167,7 @@ "LANGFLOW_API_BASE", "LANGFUSE_API_BASE", "LANGGRAPH_API_BASE", + "LAYA_API_BASE", "LEMONADE_API_BASE", "LIBERTAI_API_BASE", "LINKUP_API_BASE", @@ -175,6 +180,7 @@ "MARITALK_API_BASE", "META_API_BASE", "META_LLAMA_API_BASE", + "MICROSOFT_365_COPILOT_API_BASE", "MILVUS_API_BASE", "MINIMAX_API_BASE", "MISTRAL_API_BASE", @@ -228,6 +234,7 @@ "SAIL_API_BASE", "SAMBANOVA_API_BASE", "SAP_API_BASE", + "SCALEDOWN_API_BASE", "SCALEWAY_API_BASE", "SCX_AI_API_BASE", "SCX_API_BASE",