diff --git a/CHANGELOG.md b/CHANGELOG.md index 401a1b21..0762fb90 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,81 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Upgrade notes + +- The YAML guard has new caps for every file it reads, FLUID contracts included: a file + is refused when it nests more than 128 levels deep, or when its aliases add more than + 100,000 nodes or 5,242,880 characters of text, or grow it more than 100-fold past 1,000 + nodes. The 5 MiB size cap now counts UTF-8 bytes for text as well as for bytes. ODCS and + ODPS files now go through the guard at all, so `fluid odps import`/`validate`, `fluid + odcs validate`/`import`, `fluid odps-bitol validate`, the `contractId` resolver and + `fluid forge --seed-from` also refuse a YAML file over 5 MiB or with more than 50 + aliases, the caps FLUID contracts already had. Write the repeated blocks out instead of + reusing them through anchors. A bundle directory without a readable ODPS doc that holds + a file the guard refuses, or one that does not parse, fails instead of importing as + ODCS-only. +- A file the resolver probes for a port's `contractId` must now declare that `contractId` + as its `id`. A bundle that relied on a file named after the port's `contractId` but + declaring another id fails a strict import, or skips the port under `--lenient` (an + input port is always skipped); make the port's `contractId` and the ODCS `id` agree. + +### Fixed + +- **`fluid forge --seed-from` catches a dropped expose.** When the model left out an + expose the seed declares, the ground-truth guard raised `IndexError`, logged + `structural_seed_guard_failed` and kept the generated contract, so the expose was lost + without an error. A missing expose is now a seed mismatch, which the repair loop feeds + back to the model on its next attempt. +- **A malformed or mismatched `contractId` no longer breaks an ODPS import.** A + `contractId` with a NUL byte, a value that is not a string, a segment longer than a file + name may be, or an unparsable URL such as `http://[::1` made `fluid odps import` fail + with `ValueError`, `TypeError` or `OSError`, even with `--lenient` and on input ports, + which always import leniently. It is now an unresolved contract: `--lenient` and input + ports skip it with a warning, and a strict import stops with an unresolved-contract + error (`not a valid contractId` for a NUL byte or a value that is not a string, `false` + and `0` included, which a strict import used to skip; an empty or missing one is + ignored, as before). A file the resolver probes + for a `contractId`, such as `orders.odcs.yaml` for `commerce.orders`, is now used only + when it declares that `id`; a file that declares another contract is skipped with a + warning naming both ids instead of being linked to the port. +- **`fluid odps import ` no longer drops a product doc it cannot read.** A product + file with a YAML syntax error, or with an upper-case suffix such as `PRODUCT.ODPS.YAML` + (which the ODCS reader parsed as JSON), was skipped without a word, and the directory + imported as ODCS-only. Every ODCS/ODPS reader now reads an upper-case `.YAML`/`.YML` + suffix as YAML, and a directory file that does not parse is skipped with a warning + naming it; a directory left without a readable ODPS doc fails. +- **`fluid generate artifacts` reports a contract it cannot read.** With `--emit + schedule` alone, a contract that did not parse skipped the only emitter and exited 0 + with no files; a contract that was not UTF-8 or not a mapping ended in "Unexpected + error". Each is now a `generate_artifacts_failed` error naming the file. +- **`fluid odcs validate --report` writes the report when the YAML does not parse.** The + command printed the failure and exited 1 without the report, so a CI gate reading it + found no file; the report now records a failed `parse` pass. + +### Security + +- **The YAML guard caps what aliases expand to, in nodes and in text, and how deep a file + nests.** It allowed 50 aliases, which is not a bound: 22 levels of `[*a, *a]` are 44 + aliases that expand to 25,165,822 nodes, and a 685-byte contract shaped like that held + `fluid validate` for more than 90 seconds. A document is now also refused when its + aliases add more than 100,000 nodes or 5,242,880 characters of text (a + 1,000,000-character string repeated through two nested lists of four aliases adds + 20,000,000), or grow it more than 100-fold once it is past 1,000 nodes, and the error + names the sizes. A file nested more than 128 levels deep is refused before PyYAML scans + it to the end (a 64 KB file nested 32,000 deep took over 40 seconds) or fails with + `RecursionError`, and a stream is read only up to the 5 MiB cap. The ODCS and ODPS + readers had no cap at all: `fluid odps import`, the `contractId` resolver (fetched + contracts included), `fluid forge --seed-from`, `fluid odps-bitol validate` and `fluid + odcs validate`/`import` read YAML with a plain `yaml.safe_load`. They now use the same + guard and report a refused file as a failed validation or import, and the OpenAPI + validator, the artifact fan-out and the federation manifest reader report it as the + unreadable file it is instead of failing with `UnsafeYamlError`. +- **A committed `.env` can no longer point `REKA_API_BASE` at the FLUID LLM key.** + litellm 1.106.0 added Reka to its registry of OpenAI-compatible providers, read + from `REKA_API_BASE`, and the vendored list of litellm endpoints did not have it. + With `FLUID_LLM_API_KEY` (or the keyring key) in use, a project's `REKA_API_BASE` + is now ignored, as every other litellm provider endpoint is. + ## [0.20.0] β€” 2026-10-15 ### Security advisories diff --git a/fluid_build/cli/_common.py b/fluid_build/cli/_common.py index 78134d6a..3d7dca7b 100644 --- a/fluid_build/cli/_common.py +++ b/fluid_build/cli/_common.py @@ -97,14 +97,18 @@ def load_yaml_json(path: Path) -> Any: ``if .yaml then yaml.safe_load else json.load`` pattern that was previously duplicated across ``cli/opds.py``, ``cli/odcs.py``, and ``forge/core/artifact_fanout.py``. + + YAML goes through :func:`fluid_build.util.safe_yaml.load_yaml_safe`: the + files it reads (ODCS contracts for ``fluid odcs``) come from other teams, + so one over its caps raises ``UnsafeYamlError``. """ import json as _json - with open(path, encoding="utf-8") as f: - if path.suffix in (".yaml", ".yml"): - import yaml as _yaml + from fluid_build.util.safe_yaml import load_yaml_safe - return _yaml.safe_load(f) + with open(path, encoding="utf-8") as f: + if path.suffix.lower() in (".yaml", ".yml"): + return load_yaml_safe(f) return _json.load(f) diff --git a/fluid_build/cli/forge_copilot_seed.py b/fluid_build/cli/forge_copilot_seed.py index fb97b021..5b01c8c0 100644 --- a/fluid_build/cli/forge_copilot_seed.py +++ b/fluid_build/cli/forge_copilot_seed.py @@ -204,13 +204,22 @@ def _ground_truth_paths_for(fluid: Mapping[str, Any]) -> List[str]: def resolve_at_path(fluid: Mapping[str, Any], path: str) -> Any: - """Look up the value at a dotted/indexed path like ``exposes[0].contract.schema``.""" + """Look up the value at a dotted/indexed path like ``exposes[0].contract.schema``. + + Returns ``None`` for anything missing, including an index past the end of + a list or into a non-list. The candidate is LLM output of any shape, and a + lookup that raised made :func:`diff_against_seed` raise, which the runtime + logs as ``structural_seed_guard_failed`` and then keeps the contract. + """ import re cursor: Any = fluid for token in re.findall(r"[^.\[\]]+|\[\d+\]", path): if token.startswith("[") and token.endswith("]"): - cursor = cursor[int(token[1:-1])] + index = int(token[1:-1]) + if not isinstance(cursor, (list, tuple)) or index >= len(cursor): + return None + cursor = cursor[index] else: if isinstance(cursor, Mapping): cursor = cursor.get(token) diff --git a/fluid_build/cli/odcs.py b/fluid_build/cli/odcs.py index 7e2086b0..d154777d 100644 --- a/fluid_build/cli/odcs.py +++ b/fluid_build/cli/odcs.py @@ -433,9 +433,17 @@ def _run_odcs_import(args): import json from pathlib import Path + import yaml + + from fluid_build.util.safe_yaml import UnsafeYamlError + # Import ODCS provider = OdcsProvider() - fluid_contract = provider.import_contract(args.odcs_file) + try: + fluid_contract = provider.import_contract(args.odcs_file) + except (UnsafeYamlError, yaml.YAMLError) as exc: + console_error(f"Error importing: {exc}") + return 1 # Generate output path if not specified if not args.output: @@ -506,9 +514,10 @@ def _run_odcs_validate(args): Exit code is 1 iff any pass surfaces a violation. With ``--report PATH`` every result is also serialised to a JSON file for CI gates. """ - import json from pathlib import Path + import yaml + from fluid_build.cli._common import load_yaml_json from fluid_build.providers.odcs.validation import ( collect_errors, @@ -516,15 +525,22 @@ def _run_odcs_validate(args): roundtrip_check, validate_via_vowl, ) + from fluid_build.util.safe_yaml import UnsafeYamlError file_path = Path(args.odcs_file) - odcs_data = load_yaml_json(file_path) - report: dict = { "file": str(file_path), "passes": {}, "ok": True, } + try: + odcs_data = load_yaml_json(file_path) + except (UnsafeYamlError, yaml.YAMLError) as exc: + cprint(f"βœ— Validation failed: {exc}") + report["passes"]["parse"] = {"ok": False, "error": str(exc)[:300]} + report["ok"] = False + _write_validate_report(args, report) + return 1 # Pass 1 β€” jsonschema (multi-error) schema = load_schema() @@ -590,15 +606,18 @@ def _run_odcs_validate(args): f"extra={len(rt['extra'])} changed={len(rt['changed'])}" ) - # Optional structured report (CI gate) + _write_validate_report(args, report) + return 0 if report["ok"] else 1 + + +def _write_validate_report(args, report: dict) -> None: + """Write the optional structured report (CI gate) for ``--report``.""" if getattr(args, "report", None): report_path = Path(args.report) report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(report, indent=2, default=str), encoding="utf-8") cprint(f"πŸ“„ Report: {report_path}") - return 0 if report["ok"] else 1 - def _run_odcs_info(args): """Run ODCS info command.""" diff --git a/fluid_build/cli/odps_standard.py b/fluid_build/cli/odps_standard.py index 7ead6f47..7938de35 100644 --- a/fluid_build/cli/odps_standard.py +++ b/fluid_build/cli/odps_standard.py @@ -137,10 +137,10 @@ def validate_command(odps_file: str): file_path = Path(odps_file) with open(file_path, encoding="utf-8") as f: - if file_path.suffix in (".yaml", ".yml"): - import yaml + if file_path.suffix.lower() in (".yaml", ".yml"): + from fluid_build.util.safe_yaml import load_yaml_safe - odps_data = yaml.safe_load(f) + odps_data = load_yaml_safe(f) else: odps_data = json.load(f) @@ -319,12 +319,18 @@ def _run_odps_validate(args): import json from pathlib import Path + import yaml + + from fluid_build.util.safe_yaml import UnsafeYamlError, load_yaml_safe + file_path = Path(args.odps_file) with open(file_path, encoding="utf-8") as f: - if file_path.suffix in (".yaml", ".yml"): - import yaml - - odps_data = yaml.safe_load(f) + if file_path.suffix.lower() in (".yaml", ".yml"): + try: + odps_data = load_yaml_safe(f) + except (UnsafeYamlError, yaml.YAMLError) as exc: + cprint(f"βœ— Validation failed: {exc}") + return 1 else: odps_data = json.load(f) diff --git a/fluid_build/credentials/litellm_endpoints.py b/fluid_build/credentials/litellm_endpoints.py index 31968a29..67af1e4c 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 and 1.105.0 wheels -(FLUID's floor, the last release Python 3.10 can import, and the newest), as -the union of: +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: * ``_API_BASE`` for every ``litellm.provider_list`` entry (``-`` read as ``_``); @@ -218,6 +218,7 @@ "RAGFLOW_API_BASE", "RECRAFT_API_BASE", "REDUCTO_API_BASE", + "REKA_API_BASE", "REPLICATE_API_BASE", "RUNWAYML_API_BASE", "S3_VECTORS_API_BASE", diff --git a/fluid_build/forge/core/artifact_fanout.py b/fluid_build/forge/core/artifact_fanout.py index 7def0a79..45865461 100644 --- a/fluid_build/forge/core/artifact_fanout.py +++ b/fluid_build/forge/core/artifact_fanout.py @@ -86,7 +86,7 @@ import yaml from fluid_build.forge.core.bundle import _slug, build_manifest, validate_manifest -from fluid_build.util.safe_yaml import load_yaml_safe +from fluid_build.util.safe_yaml import UnsafeYamlError, load_yaml_safe LOG = logging.getLogger("fluid.forge.core.artifact_fanout") @@ -258,7 +258,9 @@ def _contract_is_reference_only(contract_path: Path) -> bool: try: with open(contract_path, "r", encoding="utf-8") as fh: contract = load_yaml_safe(fh) or {} - except (OSError, yaml.YAMLError): + except (OSError, UnicodeDecodeError, yaml.YAMLError, UnsafeYamlError): + return False + if not isinstance(contract, dict): return False builds = contract.get("builds") if not isinstance(builds, list): @@ -277,8 +279,8 @@ def _load_schedule_contract( Loaded through the same loader (overlay, ``$ref`` resolution and alias normalisation included) so the gate below and the renderer agree on which builds carry a trigger. ``None`` only when the contract file itself - cannot be read, which skips ``schedule`` exactly as it always has (the - other emitters then report the unreadable file). A contract that reads + cannot be read; ``run_fanout`` then reports the unreadable file, as every + other emitter does. A contract that reads but will not load, a malformed overlay above all, raises: skipping would drop the schedule while stage 3 reported success, and ``fluid apply --env`` fails on the same overlay. @@ -288,7 +290,7 @@ def _load_schedule_contract( try: with open(contract_path, "r", encoding="utf-8") as fh: load_yaml_safe(fh) - except (OSError, yaml.YAMLError): + except (OSError, UnicodeDecodeError, yaml.YAMLError, UnsafeYamlError): return None try: contract = load_contract_with_overlay(str(contract_path), overlay_env, logger) @@ -459,8 +461,17 @@ def _slug_from_contract(contract: Dict[str, Any]) -> str: def _load_contract(contract_path: Path) -> Dict[str, Any]: - with open(contract_path, "r", encoding="utf-8") as fh: - doc = load_yaml_safe(fh) + try: + with open(contract_path, "r", encoding="utf-8") as fh: + doc = load_yaml_safe(fh) + except OSError as exc: + # FanoutError is what `fluid generate artifacts` reports; anything else + # ends as an unhandled exception. + raise FanoutError( + f"contract at {contract_path} could not be read: {exc}", key=None + ) from exc + except (UnicodeDecodeError, yaml.YAMLError, UnsafeYamlError) as exc: + raise FanoutError(f"contract at {contract_path} did not parse: {exc}", key=None) from exc if not isinstance(doc, dict): raise FanoutError( f"contract at {contract_path} did not parse as a mapping", @@ -888,6 +899,10 @@ def run_fanout( if "schedule" in emits: schedule_contract = _load_schedule_contract(resolved_contract, overlay_env, logger) schedule_skip = _schedule_skip_reason(schedule_contract) + if schedule_skip == "generate_artifacts_skip_schedule_unreadable": + # Raises the FanoutError every other emitter would; with + # `--emit schedule` alone, skipping would report success. + _load_contract(resolved_contract) if "schedule" in emits and schedule_skip is not None: logger.info(schedule_skip, extra={"hint": _SCHEDULE_SKIP_HINTS[schedule_skip]}) emits = [k for k in emits if k != "schedule"] diff --git a/fluid_build/forge/core/artifact_validators.py b/fluid_build/forge/core/artifact_validators.py index fda4222b..5577ffd9 100644 --- a/fluid_build/forge/core/artifact_validators.py +++ b/fluid_build/forge/core/artifact_validators.py @@ -285,8 +285,8 @@ def _validate_against_schema( # Parse YAML or JSON depending on extension. # SECURITY (billion-laughs DoS): generated/vendored artifacts are - # untrusted content β€” route YAML through ``load_yaml_safe`` (50-alias + - # 5 MiB caps) instead of bare ``yaml.safe_load`` so a hostile anchor- + # untrusted content β€” route YAML through ``load_yaml_safe`` (size, alias + # and alias-expansion caps) instead of bare ``yaml.safe_load`` so a hostile anchor- # expansion payload can't OOM the artifact validator. from fluid_build.util.safe_yaml import UnsafeYamlError, load_yaml_safe diff --git a/fluid_build/forge/core/validation.py b/fluid_build/forge/core/validation.py index 8a1537ac..0fcd4a5e 100644 --- a/fluid_build/forge/core/validation.py +++ b/fluid_build/forge/core/validation.py @@ -442,8 +442,8 @@ def _validate_yaml_file(self, file_path: Path): # validator before the cap ever ran. Stat FIRST and reject # oversized files as a validation issue β€” mirrors the # stat-before-read in ``forge/federation.py`` and reuses the same - # :data:`MAX_YAML_BYTES` ceiling. ``load_yaml_safe`` (50-alias + - # 5 MiB caps) remains the parse path (defence in depth). Falls + # :data:`MAX_YAML_BYTES` ceiling. ``load_yaml_safe`` (size, alias + # and alias-expansion caps) remains the parse path (defence in depth). Falls # back to the bare loader only when PyYAML itself is unavailable. try: from fluid_build.util.safe_yaml import MAX_YAML_BYTES, load_yaml_safe diff --git a/fluid_build/forge/core/validators.py b/fluid_build/forge/core/validators.py index e643efdd..83472ccf 100644 --- a/fluid_build/forge/core/validators.py +++ b/fluid_build/forge/core/validators.py @@ -51,7 +51,7 @@ confine_ref, iter_external_refs, ) -from fluid_build.util.safe_yaml import load_yaml_safe +from fluid_build.util.safe_yaml import UnsafeYamlError, load_yaml_safe LOG = logging.getLogger("fluid.forge.core.validators") @@ -307,7 +307,7 @@ def validate_openapi( spec = json.loads(content.decode("utf-8")) else: spec = load_yaml_safe(content) - except (yaml.YAMLError, json.JSONDecodeError) as exc: + except (yaml.YAMLError, UnsafeYamlError, json.JSONDecodeError) as exc: return [ ValidationIssue( file=path, diff --git a/fluid_build/forge/federation.py b/fluid_build/forge/federation.py index 5cb6b984..76df5e10 100644 --- a/fluid_build/forge/federation.py +++ b/fluid_build/forge/federation.py @@ -634,7 +634,7 @@ def load_federation_manifest(workspace_root: Path) -> FederationManifest: try: with path.open(encoding="utf-8") as f: doc = load_yaml_safe(f) or {} - except (OSError, yaml.YAMLError) as exc: + except (OSError, yaml.YAMLError, UnsafeYamlError) as exc: LOG.warning("federation_manifest_unreadable: path=%s error=%s", path, exc) return FederationManifest() rows = doc.get("workspaces", []) diff --git a/fluid_build/providers/odcs/io.py b/fluid_build/providers/odcs/io.py index d15df450..8a9b68ea 100644 --- a/fluid_build/providers/odcs/io.py +++ b/fluid_build/providers/odcs/io.py @@ -44,10 +44,12 @@ def read_input(path: Union[str, Path]) -> Dict[str, Any]: """Load an ODCS YAML/JSON file. - Routes ODCS-shaped files through ``OpenDataContractStandard.from_file()`` - so Pydantic catches structural issues at the boundary. Falls back to - plain ``yaml.safe_load`` / ``json.load`` when the file doesn't have - the ``apiVersion`` + ``kind`` markers (e.g. partial fixtures). + YAML is read with :func:`fluid_build.util.safe_yaml.load_yaml_safe`, so a + file over its size, alias-count or alias-expansion caps raises + ``UnsafeYamlError`` (a ``ValueError``). ODCS-shaped files are then routed + through ``OpenDataContractStandard.from_file()`` so Pydantic catches + structural issues at the boundary; the raw load is returned when the file + doesn't have the ``apiVersion`` + ``kind`` markers (e.g. partial fixtures). """ input_path = Path(path) raw = _raw_load(input_path) @@ -114,10 +116,12 @@ def write_output(data: Dict[str, Any], path: Union[Path, str], fmt: str) -> None def _raw_load(path: Path) -> Any: with open(path, encoding="utf-8") as f: - if path.suffix in (".yaml", ".yml"): - import yaml + if path.suffix.lower() in (".yaml", ".yml"): + # ODCS / ODPS files come from other teams and catalogs, so they + # get the anchor-expansion caps any untrusted YAML gets. + from fluid_build.util.safe_yaml import load_yaml_safe - return yaml.safe_load(f) + return load_yaml_safe(f) return json.load(f) diff --git a/fluid_build/providers/odps_standard/provider.py b/fluid_build/providers/odps_standard/provider.py index 3d94f431..bf563fdb 100644 --- a/fluid_build/providers/odps_standard/provider.py +++ b/fluid_build/providers/odps_standard/provider.py @@ -28,7 +28,9 @@ import os from collections.abc import Iterable, Mapping, Sequence from pathlib import Path -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Tuple, Union + +import yaml from fluid_build.providers._path_safety import contained_path, safe_filename_stem from fluid_build.providers.base import ApplyResult, BaseProvider, ProviderError @@ -294,25 +296,23 @@ def import_directory( return a FLUID contract with one expose per ODCS file and no product wrapper. - Raises if the directory contains zero or more than one ODPS doc. + Raises if the directory contains more than one ODPS doc, or none + while a file that may be it could not be read (it did not parse, or + the YAML caps refused it). Such a file next to a readable ODPS doc + is skipped with a warning. """ root = Path(dir_path) if not root.is_dir(): raise ProviderError(f"import_directory: {root} is not a directory") - odps_files: List[Path] = [] - odcs_files: List[Path] = [] - for child in sorted(root.rglob("*")): - if not child.is_file(): - continue - suf = child.suffix.lower() - if suf not in (".yaml", ".yml", ".json"): - continue - kind = _sniff_kind(child) - if kind == "DataProduct": - odps_files.append(child) - elif kind == "DataContract": - odcs_files.append(child) + odps_files, odcs_files, unreadable = _classify_bundle_files(root, self.logger) + if unreadable and not odps_files: + # An unreadable file may be the product: importing the rest as an + # ODCS-only directory would drop it without an error. + raise ProviderError( + f"import_directory: found no readable ODPS doc in {root}, and could not " + f"read {len(unreadable)} file(s) that may be it: {unreadable}" + ) if len(odps_files) > 1: raise ProviderError( @@ -389,7 +389,7 @@ def _resolve_and_merge_ports( if stub is None: continue cid = port.get("contractId") - if not cid: + if cid is None or cid == "": continue resolved = self._safe_resolve(resolver, cid, lenient=lenient) if resolved is None: @@ -406,7 +406,7 @@ def _resolve_and_merge_ports( if stub is None: continue cid = port.get("contractId") - if not cid: + if cid is None or cid == "": continue # Always lenient for input ports β€” upstream contracts may live elsewhere resolved = self._safe_resolve(resolver, cid, lenient=True) @@ -421,7 +421,8 @@ def _safe_resolve( return resolver.resolve(contract_id) except ContractNotFound as exc: if lenient: - self.logger.warning("Skipping unresolved contract %s: %s", contract_id, exc) + # %r: contractId is document text and can carry a NUL byte. + self.logger.warning("Skipping unresolved contract %r: %s", contract_id, exc) return None raise @@ -510,12 +511,29 @@ def validate_product(self, odps: Mapping[str, Any]) -> None: validate(odps, schema) -def _sniff_kind(path: Path) -> Optional[str]: - """Return ``kind`` from a YAML/JSON file without raising.""" - try: - data = read_input(path) - except Exception: - return None - if not isinstance(data, Mapping): - return None - return data.get("kind") +def _classify_bundle_files( + root: Path, logger: logging.Logger +) -> Tuple[List[Path], List[Path], List[str]]: + """Sort a directory's YAML/JSON files into ODPS docs, ODCS contracts and + files that could not be read, which are skipped with a warning.""" + odps_files: List[Path] = [] + odcs_files: List[Path] = [] + unreadable: List[str] = [] + for child in sorted(root.rglob("*")): + if not child.is_file() or child.suffix.lower() not in (".yaml", ".yml", ".json"): + continue + # A file that does not parse, or that the YAML caps refuse, may be the + # product doc, so it is reported, not mistaken for an unrelated file. + try: + data = read_input(child) + # ValueError: UnsafeYamlError, UnicodeDecodeError, JSONDecodeError. + except (OSError, ValueError, RecursionError, yaml.YAMLError) as exc: + logger.warning("Skipping %r: %s", str(child), exc) + unreadable.append(str(child)) + continue + kind = data.get("kind") if isinstance(data, Mapping) else None + if kind == "DataProduct": + odps_files.append(child) + elif kind == "DataContract": + odcs_files.append(child) + return odps_files, odcs_files, unreadable diff --git a/fluid_build/providers/odps_standard/resolver.py b/fluid_build/providers/odps_standard/resolver.py index 403fcfdf..54b6246f 100644 --- a/fluid_build/providers/odps_standard/resolver.py +++ b/fluid_build/providers/odps_standard/resolver.py @@ -17,7 +17,8 @@ Resolution order (per the Phase 3 plan): 1. **local file** β€” probe candidate filenames in ``base_path`` and a few - conventional subdirs. + conventional subdirs. A probed file is used only when it declares the + ``contractId`` as its ``id``; one that declares another is skipped. 2. **http(s) URL** β€” only when the ``contractId`` is itself a URL (or a ``hint`` URL is given) and ``allow_remote=True``. @@ -56,6 +57,7 @@ from fluid_build.util.safe_http import ( fetch_bytes as _fetch_bytes, ) +from fluid_build.util.safe_yaml import UnsafeYamlError, load_yaml_safe LOG = logging.getLogger(__name__) # Probing matrix sized for the common case (Bitol fragments layout: ODPS doc @@ -75,10 +77,12 @@ class ContractNotFound(ProviderError): _MAX_TRIED_DISPLAYED = 5 - def __init__(self, contract_id: str, tried: List[str]): + def __init__(self, contract_id: str, tried: List[str], *, reason: Optional[str] = None): self.contract_id = contract_id self.tried = list(tried) - if tried: + if reason: + msg = f"Could not resolve contractId {contract_id!r}: {reason}" + elif tried: head = tried[: self._MAX_TRIED_DISPLAYED] tail = ( f"\n …and {len(tried) - self._MAX_TRIED_DISPLAYED} more" @@ -177,6 +181,10 @@ def index_directory(self, dir_path: Union[str, Path]) -> None: def resolve( self, contract_id: str, *, hint: Optional[Union[str, Path]] = None ) -> ResolvedContract: + # Report an id that names no file and no URL as not found, so lenient + # imports skip the port; the path probes used to raise on it instead. + if not _is_valid_contract_id(contract_id): + raise ContractNotFound(contract_id, [], reason=_INVALID_CONTRACT_ID) if contract_id in self._cache: return self._cache[contract_id] @@ -197,13 +205,9 @@ def resolve( return self._cache_and_return(contract_id, resolved) # 2. Local file probes - if self.base_path is not None: - for candidate in self._local_candidates(contract_id): - tried.append(str(candidate)) - if candidate.exists(): - resolved = self._load_local(contract_id, candidate) - if resolved is not None: - return self._cache_and_return(contract_id, resolved) + resolved = self._probe_local(contract_id, tried) + if resolved is not None: + return self._cache_and_return(contract_id, resolved) # 3. http(s) if the contractId itself is a URL if _looks_like_url(contract_id): @@ -227,6 +231,20 @@ def resolve( # ---- internals ------------------------------------------------------ + def _probe_local(self, contract_id: str, tried: List[str]) -> Optional[ResolvedContract]: + """The first probed file under ``base_path`` that declares ``contract_id``.""" + for candidate in self._local_candidates(contract_id): + tried.append(str(candidate)) + try: + present = candidate.exists() + except OSError: # ENAMETOOLONG: a path segment past the OS limit + continue + if present: + resolved = self._load_local(contract_id, candidate, require_id=True) + if resolved is not None: + return resolved + return None + def _local_candidates(self, contract_id: str) -> List[Path]: if self.base_path is None: return [] @@ -247,7 +265,7 @@ def _local_candidates(self, contract_id: str) -> List[Path]: candidate = sub_dir / f"{stem}{ext}" try: resolved = candidate.resolve() - except (OSError, RuntimeError): + except (OSError, ValueError, RuntimeError): # NUL byte, symlink loop continue if not _is_within(resolved, base_resolved): continue @@ -296,7 +314,9 @@ def _index_file(self, path: Path) -> None: return self._index.setdefault(odcs_id, path) - def _load_local(self, contract_id: str, path: Path) -> Optional[ResolvedContract]: + def _load_local( + self, contract_id: str, path: Path, *, require_id: bool = False + ) -> Optional[ResolvedContract]: try: data = read_input(path) except Exception as exc: @@ -304,6 +324,17 @@ def _load_local(self, contract_id: str, path: Path) -> Optional[ResolvedContract return None if not isinstance(data, Mapping): return None + # A probe only guesses the file from the contractId: the last-segment + # probe tries orders.odcs.yaml for "commerce.orders". The file is that + # contract only if it declares that id, as ODPS ports reference ODCS ids. + if require_id and data.get("id") != contract_id: + LOG.warning( + "Skipping %r: it declares id %r, not contractId %r", + str(path), # %r: the bundle names its files, control characters included + data.get("id"), + contract_id, + ) + return None self._validate(contract_id, data, origin=str(path)) return ResolvedContract( odcs=dict(data), @@ -343,7 +374,10 @@ def _load_remote(self, contract_id: str, url: str) -> Optional[ResolvedContract] try: data = json.loads(text) except json.JSONDecodeError: - data = yaml.safe_load(text) + data = load_yaml_safe(text) + except UnsafeYamlError as exc: + # Its message carries only sizes and counts, never body text. + raise ContractNotFound(contract_id, [url], reason=str(exc)) from None except (UnicodeDecodeError, yaml.YAMLError): # Don't propagate parser exception text β€” it can include body fragments raise ContractNotFound(contract_id, [url]) from None @@ -383,8 +417,23 @@ def _cache_and_return(self, contract_id: str, resolved: ResolvedContract) -> Res return resolved +_INVALID_CONTRACT_ID = "not a valid contractId (expected a non-empty string with no NUL bytes)" + + +def _is_valid_contract_id(value: object) -> bool: + """True for a contractId the probes can use: a non-empty string without NUL. + + contractId is document text, so anything else (a number, a list, a NUL + byte that ``Path.resolve`` rejects) names no file and no URL. + """ + return isinstance(value, str) and bool(value) and "\x00" not in value + + def _looks_like_url(value: str) -> bool: - parsed = urlparse(value) + try: + parsed = urlparse(value) + except ValueError: # e.g. "http://[::1": urlparse rejects the unclosed IPv6 bracket + return False return parsed.scheme in ("http", "https") and bool(parsed.netloc) diff --git a/fluid_build/util/safe_yaml.py b/fluid_build/util/safe_yaml.py index 2419a113..1509504e 100644 --- a/fluid_build/util/safe_yaml.py +++ b/fluid_build/util/safe_yaml.py @@ -15,15 +15,38 @@ """Bounded YAML loading for untrusted input. ``yaml.safe_load`` blocks arbitrary-object construction but does NOT cap -anchor/alias expansion: a small document with nested YAML aliases can -balloon to gigabytes when composed (the "billion laughs" denial-of-service). +anchor/alias expansion: a small document with nested YAML aliases loads as +a few shared objects, then balloons to gigabytes in whatever walks it next, +validation, error messages, ``json.dumps`` (the "billion laughs" +denial-of-service). There is no drop-in OSS parser that fixes this without dropping anchor support entirely (StrictYAML) β€” see PyYAML issue #235. So we pre-scan the cheap event stream: ``yaml.parse`` tokenises the document WITHOUT resolving aliases, so it stays linear-time even for a malicious payload. We count -``AliasEvent``s and reject over-budget input before the expensive -``safe_load`` compose/construct step ever runs. +``AliasEvent``s and reject over-budget input before anything is composed. + +The count alone does not bound expansion: 22 levels of +``l: &l [*l, *l]`` are 44 aliases that expand to 25,165,822 +nodes. So a document that has aliases is then composed (composition keeps +an alias as a second reference to its anchor's node, so the graph stays as +small as the text) and walked once, charging every alias the size of the +node it repeats, in nodes and in scalar characters, before anything is +constructed: one anchored multi-megabyte string repeated through a few nested +lists adds few nodes but gigabytes of text. The node budget follows OmegaConf +2.4's YAML alias limits (``omegaconf/_yaml.py``, BSD-3-Clause) with two +differences. It caps what aliases ADD, not the document's total, so a large +contract without aliases loads. And a recursive alias is not refused here: it +is counted once. The FLUID contract loader and the contract API refuse such a +document (``contract_load_failed``); the ODCS/ODPS validators and importers do +not, and can report it valid, write it back out with its anchors, or fail when +serialising it to JSON. + +The pre-scan also stops at a nesting depth of :data:`MAX_YAML_DEPTH`, the +default recursion limit of ``serde_json``. PyYAML's scanner slows with flow +depth (a 64 KB file nested 32,000 deep took over 40 seconds to scan), and its +recursive composer fails with ``RecursionError`` a few hundred levels down, +fewer the deeper the caller's own stack is. Use :func:`load_yaml_safe` for any YAML that arrives from outside the process: LLM output, network/catalog responses, and bundle contents. @@ -40,6 +63,19 @@ # dozens of nested alias references to reach dangerous expansion. MAX_YAML_BYTES = 5 * 1024 * 1024 MAX_YAML_ALIASES = 50 +# 14x the deepest YAML file in this repo (9 levels), and short of the depth +# where PyYAML's composer exhausts the recursion limit (about 490 levels from +# an empty stack), so a deep file is refused instead of crashing. +MAX_YAML_DEPTH = 128 +# Nodes that aliases may add to the walked document, and how many times its +# size they may make it once it is past _ALIAS_RATIO_MIN_NODES (OmegaConf's +# 100x over 1,000 nodes). Reusing an anchored block adds far less. +MAX_YAML_ALIAS_EXPANSION = 100_000 +_ALIAS_EXPANSION_RATIO = 100 +_ALIAS_RATIO_MIN_NODES = 1_000 +# Scalar characters aliases may add: at most the input cap again, so what a +# consumer walks or writes out stays within about twice the file it read. +MAX_YAML_ALIAS_EXPANSION_CHARS = MAX_YAML_BYTES class UnsafeYamlError(ValueError): @@ -51,33 +87,158 @@ def load_yaml_safe( *, max_bytes: int = MAX_YAML_BYTES, max_aliases: int = MAX_YAML_ALIASES, + max_alias_expansion: int = MAX_YAML_ALIAS_EXPANSION, + max_alias_expansion_chars: int = MAX_YAML_ALIAS_EXPANSION_CHARS, + max_depth: int = MAX_YAML_DEPTH, ) -> Any: """``yaml.safe_load`` with billion-laughs (anchor-expansion) protection. - Accepts a ``str``, ``bytes``, or a readable stream. ``yaml.parse`` - does not resolve aliases, so the pre-scan is cheap even for a hostile - payload β€” ``AliasEvent``s are counted and the document is rejected - before the expensive ``safe_load`` compose/construct step. + Accepts a ``str``, ``bytes``, or a readable stream; ``max_bytes`` counts + UTF-8 bytes for all three. ``yaml.parse`` does not resolve aliases, so the + pre-scan is cheap even for a hostile payload β€” ``AliasEvent``s are counted + and nesting past ``max_depth`` is refused before the expensive + ``safe_load`` compose/construct step. A document with aliases is then + composed and refused if its aliases add more than ``max_alias_expansion`` + nodes or ``max_alias_expansion_chars`` scalar characters, or grow it past + the ratio cap. """ if source is None: return None + source = _bounded_text(source, max_bytes) + if _count_aliases(source, max_aliases, max_depth) == 0: + return yaml.safe_load(source) + loader = yaml.SafeLoader(source) + try: + node = loader.get_single_node() + if node is None: + return None + _check_alias_expansion(node, max_alias_expansion, max_alias_expansion_chars) + return loader.construct_document(node) + finally: + loader.dispose() + + +def _bounded_text(source: Any, max_bytes: int) -> str: + """``source`` as text, refused when it is over ``max_bytes`` UTF-8 bytes.""" if hasattr(source, "read"): - source = source.read() + # One unit past the cap is enough to refuse a stream; reading a huge + # file whole just to measure it is the exhaustion the cap prevents. + source = _read_bounded(source, max_bytes + 1) + if isinstance(source, (bytes, str)) and _utf8_size(source) > max_bytes: + raise UnsafeYamlError(f"YAML input is over the {max_bytes}-byte safety cap") if isinstance(source, bytes): source = source.decode("utf-8") if not isinstance(source, str): raise UnsafeYamlError(f"load_yaml_safe expects text, got {type(source).__name__}") - if len(source) > max_bytes: - raise UnsafeYamlError( - f"YAML input is {len(source)} bytes, over the {max_bytes}-byte safety cap" - ) + return source + + +def _count_aliases(source: str, max_aliases: int, max_depth: int) -> int: + """Alias references in ``source``, from the event stream (which resolves + none), refused past ``max_aliases`` or nesting past ``max_depth``.""" aliases = 0 + depth = 0 for event in yaml.parse(source, Loader=yaml.SafeLoader): - if isinstance(event, yaml.AliasEvent): + if isinstance(event, yaml.CollectionStartEvent): + depth += 1 + if depth > max_depth: + raise UnsafeYamlError(f"YAML input nests more than {max_depth} levels deep") + elif isinstance(event, yaml.CollectionEndEvent): + depth -= 1 + elif isinstance(event, yaml.AliasEvent): aliases += 1 if aliases > max_aliases: raise UnsafeYamlError( f"YAML input has more than {max_aliases} alias references β€” " "possible anchor-expansion (billion-laughs) DoS" ) - return yaml.safe_load(source) + return aliases + + +def _read_bounded(stream: Any, limit: int) -> Any: + """Read at most ``limit`` units, past short reads: a raw or unbuffered + stream can return fewer than asked before its end.""" + first = stream.read(limit) + if not first or len(first) >= limit: + return first + parts = [first] + got = len(first) + while got < limit: + chunk = stream.read(limit - got) + if not chunk: + break + parts.append(chunk) + got += len(chunk) + return first[:0].join(parts) + + +def _utf8_size(source: bytes | str) -> int: + if isinstance(source, bytes) or source.isascii(): + return len(source) + return len(source.encode("utf-8", "surrogatepass")) + + +def _check_alias_expansion(root: yaml.Node, max_added: int, max_added_chars: int) -> None: + """Refuse a composed document whose aliases repeat too much of it. + + One iterative pass over the node graph sizes each node's expanded tree, + in nodes and in scalar characters, memoised so a shared node is walked + once. An alias to a node that is still being walked (one inside its own + anchor) counts as one node and no characters, so the walk ends; the + module docstring says which consumers refuse such a document. + """ + size: dict[int, int] = {} + chars: dict[int, int] = {} + walking: set[int] = set() + unique = 0 + unique_chars = 0 + stack: list[tuple[yaml.Node, bool]] = [(root, False)] + while stack: + node, children_done = stack.pop() + key = id(node) + if children_done: + walking.discard(key) + children = _children(node) + size[key] = 1 + sum(size.get(id(child), 1) for child in children) + chars[key] = _scalar_chars(node) + sum(chars.get(id(child), 0) for child in children) + continue + if key in size or key in walking: + continue + walking.add(key) + unique += 1 + unique_chars += _scalar_chars(node) + stack.append((node, True)) + stack.extend( + (child, False) + for child in _children(node) + if id(child) not in size and id(child) not in walking + ) + total = size[id(root)] + if total - unique > max_added or ( + total > _ALIAS_RATIO_MIN_NODES and total > _ALIAS_EXPANSION_RATIO * unique + ): + raise UnsafeYamlError( + f"YAML aliases expand this document from {unique} to {total} nodes, over the " + f"cap of {max_added} added nodes or {_ALIAS_EXPANSION_RATIO}x growth β€” possible " + "anchor-expansion (billion-laughs) DoS. Write the repeated blocks out, or " + "nest fewer anchors." + ) + added_chars = chars[id(root)] - unique_chars + if added_chars > max_added_chars: + raise UnsafeYamlError( + f"YAML aliases repeat {added_chars} characters of text in this document, over " + f"the cap of {max_added_chars} β€” possible anchor-expansion (billion-laughs) DoS. " + "Write the repeated blocks out, or nest fewer anchors." + ) + + +def _scalar_chars(node: yaml.Node) -> int: + return len(node.value) if isinstance(node, yaml.ScalarNode) else 0 + + +def _children(node: yaml.Node) -> list[yaml.Node]: + if isinstance(node, yaml.SequenceNode): + return list(node.value) + if isinstance(node, yaml.MappingNode): + return [part for pair in node.value for part in pair] + return [] diff --git a/tests/forge/test_artifact_fanout.py b/tests/forge/test_artifact_fanout.py index ec622f4a..e9422b37 100644 --- a/tests/forge/test_artifact_fanout.py +++ b/tests/forge/test_artifact_fanout.py @@ -37,6 +37,7 @@ _contract_is_reference_only, _emit_policies, _expose_level_policies, + _load_contract, _schedule_skip_reason, parse_emit_set, run_fanout, @@ -207,6 +208,24 @@ def test_malformed_yaml_defaults_false(self, tmp_path): c.write_text("not: valid: yaml: [\n") assert _contract_is_reference_only(c) is False + @pytest.mark.parametrize( + "content", + [b"not: valid: yaml: [\n", b"builds: \xff\xfe\n", b"- builds\n- reference\n"], + ids=["malformed", "latin-1", "list"], + ) + def test_unreadable_contract(self, tmp_path, content): + # Not reference-only, and a FanoutError (what `fluid generate artifacts` + # reports) from the loader every emitter uses, not an unhandled exception. + c = tmp_path / "c.yaml" + c.write_bytes(content) + assert _contract_is_reference_only(c) is False + with pytest.raises(FanoutError, match="did not parse"): + _load_contract(c) + + def test_load_contract_reports_a_path_it_cannot_open(self, tmp_path): + with pytest.raises(FanoutError, match="could not be read"): + _load_contract(tmp_path) + def test_orchestration_engine_detected(self): assert _schedule_skip_reason({"orchestration": {"engine": "airflow"}}) is None @@ -307,6 +326,20 @@ def test_unknown_emit_key_errors(self, bundle_tgz, tmp_path, logger): logger=logger, ) + @pytest.mark.parametrize( + "content", [b"not: valid: yaml: [\n", b"id: \xff\xfe\n"], ids=["malformed", "latin-1"] + ) + @pytest.mark.parametrize("emit", ["schedule", None], ids=["schedule-alone", "default"]) + def test_run_fanout_reports_a_contract_that_does_not_parse( + self, tmp_path, logger, content, emit + ): + # Schedule skipped an unreadable contract for the other emitters to + # report; with `--emit schedule` alone, stage 3 exited 0 on nothing. + contract = tmp_path / "contract.fluid.yaml" + contract.write_bytes(content) + with pytest.raises(FanoutError, match="did not parse"): + run_fanout(contract, tmp_path / "art", emit_raw=emit, manifest_path=None, logger=logger) + def test_missing_bundle_errors(self, tmp_path, logger): with pytest.raises(FanoutError, match="input not found"): run_fanout( diff --git a/tests/test_forge_copilot_seed.py b/tests/test_forge_copilot_seed.py index b0fd62f6..5f9c25f2 100644 --- a/tests/test_forge_copilot_seed.py +++ b/tests/test_forge_copilot_seed.py @@ -151,6 +151,72 @@ def test_adding_unrelated_fields_does_not_trigger_diff(self) -> None: mismatches = diff_against_seed(seed, mutated) assert mismatches == [] + def test_dropped_expose_is_a_mismatch_not_a_crash(self) -> None: + # A candidate that drops a seeded expose must trip the guard. The + # lookup used to raise IndexError here; the runtime caught it as + # ``structural_seed_guard_failed`` and kept the mutated contract. + seed = load_seed(BUNDLE_DIR) + assert len(seed.fluid["exposes"]) >= 2 + mutated = copy.deepcopy(seed.fluid) + mutated["exposes"].pop() + last = len(mutated["exposes"]) + mismatches = diff_against_seed(seed, mutated) + dropped = [m for m in mismatches if m["path"].startswith(f"exposes[{last}].")] + assert dropped + assert all(m["candidate"] is None for m in dropped) + + +class TestRuntimeGuard: + def test_runtime_sends_a_dropped_expose_back_to_the_model( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # End to end through generate_copilot_artifacts with a scripted model: + # attempt 1 drops a seeded expose, attempt 2 returns the seed. Schema + # validation is stubbed (a seed is a skeleton the model completes) so + # the guard is what decides attempt 1. Streaming and the judge pass + # bypass call_llm and reach the network, so both are off. + monkeypatch.setenv("FLUID_LLM_STREAMING", "0") + monkeypatch.setenv("FLUID_COPILOT_JUDGE", "0") + import json + import logging + from types import SimpleNamespace + from unittest.mock import patch + + from fluid_build.cli import forge_copilot_runtime as rt + + seed = load_seed(BUNDLE_DIR) + dropped = copy.deepcopy(seed.fluid) + dropped["exposes"].pop() + reply = {"recommended_template": "starter", "recommended_provider": "local"} + replies = [ + json.dumps({**reply, "readme_markdown": "# p\n", "contract": contract}) + for contract in [dropped] + [seed.fluid] * 4 + ] + valid = SimpleNamespace(is_valid=True, errors=[], warnings=[]) + with ( + patch.object(rt, "call_llm", side_effect=replies), + patch.object(rt, "validate_generated_result", return_value=([], [])), + patch( + "fluid_build.schema_manager.FluidSchemaManager.validate_contract", + return_value=valid, + ), + ): + result = rt.generate_copilot_artifacts( + {"project_goal": "orders", "structural_seed": seed}, + llm_config=rt.LlmConfig( + provider="openai", + model="m", + endpoint="https://api.openai.com/v1/chat/completions", + api_key="k", + ), + discovery_report=rt.DiscoveryReport(workspace_roots=[str(BUNDLE_DIR)]), + capability_matrix={"providers": ["local"], "templates": {"starter": {}}}, + logger=logging.getLogger(__name__), + ) + first = result.attempt_reports[0].validation_errors + assert any(e.startswith("Seed ground-truth violation at exposes[1].") for e in first) + assert len(result.contract["exposes"]) == len(seed.fluid["exposes"]) + # --------------------------------------------------------------------------- # Path resolver helper @@ -164,3 +230,12 @@ def test_resolves_array_index(self) -> None: def test_returns_none_for_missing_path(self) -> None: assert resolve_at_path({}, "exposes[0].id") is None + + def test_returns_none_for_index_past_the_end(self) -> None: + assert resolve_at_path({"exposes": [{"id": "a"}]}, "exposes[1].id") is None + + @pytest.mark.parametrize("container", [{"0": {"id": "a"}}, {0: {"id": "a"}}, "ab", 7]) + def test_returns_none_for_index_into_a_non_list(self, container: object) -> None: + # An LLM can return any shape; indexing a dict, string or number is a + # missing value, not a lookup to attempt. + assert resolve_at_path({"exposes": container}, "exposes[0]") is None diff --git a/tests/test_odcs.py b/tests/test_odcs.py index e08abfea..ce8e1259 100644 --- a/tests/test_odcs.py +++ b/tests/test_odcs.py @@ -572,3 +572,65 @@ def test_info_returns_0(self): code = _run_odcs_info(args) assert code == 0 + + +# --------------------------------------------------------------------------- +# A file over the YAML caps fails cleanly (it was read with a plain safe_load) +# --------------------------------------------------------------------------- + +# 13 levels of [*a, *a]: 26 aliases that grow a 30-node file to 49,150 nodes. +_ALIAS_BOMB = "\n".join( + ["l0: &l0 [lol]"] + [f"l{i}: &l{i} [*l{i - 1}, *l{i - 1}]" for i in range(1, 14)] +) + + +@pytest.mark.parametrize( + "text, reason", + [(_ALIAS_BOMB, "aliases expand"), ("id: [unclosed\n", "Validation failed")], + ids=["alias-bomb", "malformed"], +) +def test_odcs_validate_reports_an_unreadable_file_as_a_failure( + tmp_path: Path, text: str, reason: str +) -> None: + import argparse + + contract = tmp_path / "c.yaml" + contract.write_text(text, encoding="utf-8") + report = tmp_path / "report.json" + args = argparse.Namespace( + odcs_file=str(contract), report=str(report), roundtrip=False, vowl=False + ) + with patch("fluid_build.cli.odcs.cprint") as printed: + assert _run_odcs_validate(args) == 1 + assert any(reason in c.args[0] for c in printed.call_args_list) + # A CI gate reading --report gets the failure too, not a missing file. + written = json.loads(report.read_text(encoding="utf-8")) + assert written["ok"] is False and written["passes"]["parse"]["ok"] is False + + +def test_load_yaml_json_reads_an_uppercase_suffix_as_yaml(tmp_path: Path) -> None: + from fluid_build.cli._common import load_yaml_json + + contract = tmp_path / "C.YML" + contract.write_text("kind: DataContract\n", encoding="utf-8") + assert load_yaml_json(contract) == {"kind": "DataContract"} + + +@pytest.mark.parametrize( + "text, reason", + [(_ALIAS_BOMB, "aliases expand"), ("id: [unclosed\n", "Error importing")], + ids=["alias-bomb", "malformed"], +) +def test_odcs_import_reports_an_unreadable_file_as_a_failed_import( + tmp_path: Path, text: str, reason: str +) -> None: + import argparse + + contract = tmp_path / "c.yaml" + contract.write_text(text, encoding="utf-8") + out = tmp_path / "out.yaml" + args = argparse.Namespace(odcs_file=str(contract), output=str(out), format="yaml") + with patch("fluid_build.cli.odcs.console_error") as reported: + assert _run_odcs_import(args) == 1 + assert reason in reported.call_args.args[0] + assert not out.exists() diff --git a/tests/test_odps_bitol_import.py b/tests/test_odps_bitol_import.py index f8152851..dbf81609 100644 --- a/tests/test_odps_bitol_import.py +++ b/tests/test_odps_bitol_import.py @@ -28,6 +28,7 @@ from pathlib import Path import pytest +import yaml from fluid_build.providers.base import ProviderError from fluid_build.providers.odcs import OdcsProvider @@ -39,6 +40,12 @@ BROKEN = FIXTURES / "odps" / "product-bitol-broken" +# 13 levels of [*a, *a]: 26 aliases that grow a 30-node file to 49,150 nodes. +_ALIAS_BOMB = "\n".join( + ["l0: &l0 [lol]"] + [f"l{i}: &l{i} [*l{i - 1}, *l{i - 1}]" for i in range(1, 14)] +) + + def _bundle_product_path() -> Path: return next(BUNDLE_DIR.glob("*.odps.yaml")) @@ -166,6 +173,41 @@ def test_dangling_contract_id_lenient_warns(self, provider: BitolOdpsProvider) - # The expose stub still gets created β€” just without a contract body assert len(fluid["exposes"]) == 1 + @pytest.mark.parametrize("ports, lenient", [("outputPorts", True), ("inputPorts", False)]) + def test_malformed_contract_id_skips_the_port( + self, provider: BitolOdpsProvider, ports: str, lenient: bool + ) -> None: + # A NUL byte in a contractId made the resolver raise ValueError, which + # aborted the import even with lenient=True, and even on input ports, + # which always resolve leniently. + doc = yaml.safe_load(_bundle_product_path().read_text(encoding="utf-8")) + doc.setdefault(ports, [{"name": "upstream", "version": "1.0.0"}])[0]["contractId"] = "x\x00" + fluid = provider.import_contract(doc, base_path=BUNDLE_DIR, lenient=lenient) + assert len(fluid["exposes"]) == 2 + + @pytest.mark.parametrize("contract_id", [False, 0, []], ids=["false", "zero", "empty-list"]) + def test_falsy_non_string_contract_id_fails_a_strict_import( + self, provider: BitolOdpsProvider, contract_id: object + ) -> None: + # `if not cid` skipped these, so a strict import succeeded with the port + # unresolved, while a truthy non-string failed as not a valid contractId. + doc = yaml.safe_load(_bundle_product_path().read_text(encoding="utf-8")) + doc["outputPorts"][0]["contractId"] = contract_id + with pytest.raises(ProviderError, match="not a valid contractId"): + provider.import_contract(doc, base_path=BUNDLE_DIR, lenient=False) + + def test_lenient_skip_warning_escapes_the_contract_id( + self, provider: BitolOdpsProvider + ) -> None: + from unittest.mock import patch + + from fluid_build.providers.odps_standard.resolver import ContractResolver + + with patch.object(provider, "logger") as log: + assert provider._safe_resolve(ContractResolver(), "x\x00", lenient=True) is None + fmt, contract_id, exc = log.warning.call_args.args + assert (fmt % (contract_id, exc)).startswith("Skipping unresolved contract 'x\\x00'") + def test_two_odps_docs_in_directory_raises(self, provider: BitolOdpsProvider) -> None: with pytest.raises(ProviderError, match="found 2 ODPS docs"): provider.import_directory(BROKEN / "two-odps-docs") @@ -173,3 +215,49 @@ def test_two_odps_docs_in_directory_raises(self, provider: BitolOdpsProvider) -> def test_nonexistent_directory_raises(self, provider: BitolOdpsProvider) -> None: with pytest.raises(ProviderError, match="is not a directory"): provider.import_directory(FIXTURES / "does-not-exist") + + +# Over the YAML caps, or not YAML at all. +_UNREADABLE = pytest.mark.parametrize( + "content", [_ALIAS_BOMB, "apiVersion: v1.0.0\nid: [unclosed\n"], ids=["alias-bomb", "malformed"] +) + + +class TestUnreadableBundleFile: + """A directory file that cannot be read is skipped, never silently reclassified.""" + + @_UNREADABLE + def test_unreadable_file_without_a_readable_product_fails_the_directory( + self, provider: BitolOdpsProvider, tmp_path: Path, content: str + ) -> None: + # The unreadable file may be the product: importing the rest as an + # ODCS-only directory would drop it without an error. + (tmp_path / "product.odps.yaml").write_text(content, encoding="utf-8") + for odcs in CONTRACTS_ONLY.glob("*.odcs.yaml"): + (tmp_path / odcs.name).write_text(odcs.read_text(encoding="utf-8"), encoding="utf-8") + with pytest.raises(ProviderError, match="could not read 1 file"): + provider.import_directory(tmp_path) + + @_UNREADABLE + def test_unreadable_file_next_to_a_readable_product_is_skipped( + self, provider: BitolOdpsProvider, tmp_path: Path, content: str + ) -> None: + from unittest.mock import patch + + for path in BUNDLE_DIR.iterdir(): + (tmp_path / path.name).write_text(path.read_text(encoding="utf-8"), encoding="utf-8") + (tmp_path / "docker-compose.yaml").write_text(content, encoding="utf-8") + with patch.object(provider, "logger") as log: + assert len(provider.import_directory(tmp_path)["exposes"]) == 2 + skipped = [c.args[1] for c in log.warning.call_args_list if c.args[0] == "Skipping %r: %s"] + assert skipped == [str(tmp_path / "docker-compose.yaml")] + + def test_uppercase_suffix_is_read_as_yaml( + self, provider: BitolOdpsProvider, tmp_path: Path + ) -> None: + # A case-sensitive suffix check parsed PRODUCT.ODPS.YAML as JSON, and the + # directory imported as ODCS-only, without the product's owner. + for path in BUNDLE_DIR.iterdir(): + name = path.name.upper() if path.name.endswith(".odps.yaml") else path.name + (tmp_path / name).write_text(path.read_text(encoding="utf-8"), encoding="utf-8") + assert "owner" in provider.import_directory(tmp_path) diff --git a/tests/test_odps_resolver.py b/tests/test_odps_resolver.py index 891a9e73..fb0f7dbe 100644 --- a/tests/test_odps_resolver.py +++ b/tests/test_odps_resolver.py @@ -116,6 +116,62 @@ def test_url_hint_with_remote_disabled_raises(self, tmp_path: Path) -> None: with pytest.raises(RemoteFetchDisabled): resolver.resolve("anything", hint="https://example.com/c.yaml") + @pytest.mark.parametrize( + "contract_id, reason", + [ + ("orders\x00x", "not a valid contractId"), + (123, "not a valid contractId"), + (["orders"], "not a valid contractId"), + ("", "not a valid contractId"), + ("http://[::1", "Tried"), # not a URL urlparse accepts: probed as a plain id + # A file name past the OS limit: exists() raised OSError. + pytest.param("a" * 300, "Tried", id="name-too-long"), + ], + ) + def test_malformed_contract_id_is_not_found( + self, tmp_path: Path, contract_id: object, reason: str + ) -> None: + # contractId is document text. A value the resolver cannot look up is + # a contract it did not find, which lenient imports skip; it used to + # raise ValueError/TypeError and abort the whole import. + with pytest.raises(ContractNotFound, match=reason): + ContractResolver(base_path=tmp_path).resolve(contract_id) # type: ignore[arg-type] + + +def _odcs_yaml(contract_id: str) -> str: + return ( + "apiVersion: v3.1.0\nkind: DataContract\n" + f"id: {contract_id}\nversion: 1.0.0\nstatus: active\n" + ) + + +class TestProbedIdMustMatch: + def test_probed_file_declaring_another_id_is_skipped(self, tmp_path: Path) -> None: + # The last-segment probe tries orders.odcs.yaml for "commerce.orders". + # A file there that declares a different contract must not be linked. + (tmp_path / "orders.odcs.yaml").write_text(_odcs_yaml("billing.orders"), encoding="utf-8") + resolver = ContractResolver(base_path=tmp_path, allow_remote=False) + with patch("fluid_build.providers.odps_standard.resolver.LOG") as log: + with pytest.raises(ContractNotFound): + resolver.resolve("commerce.orders") + warned = " ".join(str(a) for c in log.warning.call_args_list for a in c.args) + assert "billing.orders" in warned and "commerce.orders" in warned + + def test_probed_file_declaring_the_same_id_resolves(self, tmp_path: Path) -> None: + (tmp_path / "orders.odcs.yaml").write_text(_odcs_yaml("commerce.orders"), encoding="utf-8") + resolver = ContractResolver(base_path=tmp_path, allow_remote=False) + assert resolver.resolve("commerce.orders").odcs["id"] == "commerce.orders" + + def test_skipped_mismatch_does_not_hide_a_later_match(self, tmp_path: Path) -> None: + (tmp_path / "orders.odcs.yaml").write_text(_odcs_yaml("billing.orders"), encoding="utf-8") + (tmp_path / "contracts").mkdir() + (tmp_path / "contracts" / "orders.odcs.yaml").write_text( + _odcs_yaml("commerce.orders"), encoding="utf-8" + ) + resolver = ContractResolver(base_path=tmp_path, allow_remote=False) + result = resolver.resolve("commerce.orders") + assert Path(result.origin).parent.name == "contracts" + # --------------------------------------------------------------------------- # URL detection @@ -133,6 +189,7 @@ class TestLooksLikeUrl: ("just-an-id", False), ("product.something", False), ("https://", False), # no netloc + ("http://[::1", False), # urlparse raises on the unclosed IPv6 bracket ], ) def test_detection(self, value: str, expected: bool) -> None: @@ -144,6 +201,20 @@ def test_detection(self, value: str, expected: bool) -> None: # --------------------------------------------------------------------------- +def test_fetched_alias_bomb_is_not_found(monkeypatch: pytest.MonkeyPatch) -> None: + """A fetched body is parsed with the anchor-expansion caps any untrusted + YAML gets, not a plain yaml.safe_load.""" + from fluid_build.providers.odps_standard import resolver as resolver_mod + + lines = ["apiVersion: v3.1.0", "kind: DataContract", "l0: &l0 [lol]"] + lines += [f"l{i}: &l{i} [*l{i - 1}, *l{i - 1}]" for i in range(1, 19)] + body = "\n".join(lines).encode() + monkeypatch.setattr(resolver_mod, "_assert_safe_url", lambda url: None) + monkeypatch.setattr(resolver_mod, "_fetch_bytes", lambda *a, **k: (200, {}, body)) + with pytest.raises(ContractNotFound, match="aliases expand"): + ContractResolver(allow_remote=True).resolve("https://example.com/c.odcs.yaml") + + class TestRemoteResolution: def test_html_with_200_is_refused(self, tmp_path: Path) -> None: resolver = ContractResolver(base_path=tmp_path, allow_remote=True) diff --git a/tests/test_odps_standard.py b/tests/test_odps_standard.py index cac568af..e2dea8b2 100644 --- a/tests/test_odps_standard.py +++ b/tests/test_odps_standard.py @@ -691,5 +691,43 @@ def test_cli_group_help(self): assert "ODPS-Bitol" in result.output or "odps-bitol" in result.output.lower() +_BOMB_PRODUCT = "\n".join( + ["apiVersion: v1.0.0", "kind: DataProduct", "id: dp-001", "name: P", "status: x"] + + ["l0: &l0 [lol]"] + + [f"l{i}: &l{i} [*l{i - 1}, *l{i - 1}]" for i in range(1, 19)] +) + + +@pytest.mark.parametrize( + "text, reason", + [(_BOMB_PRODUCT, "aliases expand"), ("id: [unclosed\n", "Validation failed")], + ids=["alias-bomb", "malformed"], +) +def test_odps_validate_reports_an_unreadable_file_as_a_failure( + tmp_path: Path, text: str, reason: str +) -> None: + """The file is read with the anchor-expansion caps; one over them, or one + that is not YAML, fails the validation instead of an unhandled exception.""" + from fluid_build.cli.odps_standard import _run_odps_validate + + product = tmp_path / "product.odps.yaml" + product.write_text(text, encoding="utf-8") + + with patch("fluid_build.cli.odps_standard.cprint") as printed: + assert _run_odps_validate(argparse.Namespace(odps_file=str(product))) == 1 + assert reason in printed.call_args.args[0] + + +def test_odps_validate_reads_an_uppercase_suffix_as_yaml(tmp_path: Path) -> None: + """PRODUCT.ODPS.YAML was parsed as JSON, so the caps never ran on it.""" + from fluid_build.cli.odps_standard import _run_odps_validate + + product = tmp_path / "PRODUCT.ODPS.YAML" + product.write_text(_BOMB_PRODUCT, encoding="utf-8") + with patch("fluid_build.cli.odps_standard.cprint") as printed: + assert _run_odps_validate(argparse.Namespace(odps_file=str(product))) == 1 + assert "aliases expand" in printed.call_args.args[0] + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_safe_yaml.py b/tests/test_safe_yaml.py index 22c8baf3..97a6becb 100644 --- a/tests/test_safe_yaml.py +++ b/tests/test_safe_yaml.py @@ -18,8 +18,10 @@ from __future__ import annotations import io +from pathlib import Path import pytest +import yaml from fluid_build.util.safe_yaml import MAX_YAML_ALIASES, UnsafeYamlError, load_yaml_safe @@ -68,3 +70,223 @@ def test_alias_cap_boundary() -> None: refs = "\n".join(f"r{i}: *a" for i in range(MAX_YAML_ALIASES + 1)) with pytest.raises(UnsafeYamlError): load_yaml_safe("a: &a 1\n" + refs) + + +def _doubling_chain(levels: int) -> str: + """Each level is a list of two aliases to the level below: 2 aliases per + level, 2**levels leaves. 24 levels stay under the 50-alias cap.""" + lines = ["l0: &l0 [lol]"] + for i in range(1, levels + 1): + lines.append(f"l{i}: &l{i} [*l{i - 1}, *l{i - 1}]") + return "\n".join(lines) + + +def test_rejects_doubling_chain_under_the_alias_cap() -> None: + """48 aliases pass a count cap of 50 yet expand to 2**24 leaves (100,663,294 + nodes); a 44-alias chain like this one held `fluid validate` for over 90 s.""" + doc = _doubling_chain(24) + assert doc.count("*") <= MAX_YAML_ALIASES + with pytest.raises(UnsafeYamlError, match="expand"): + load_yaml_safe(doc) + + +def test_recursive_alias_is_left_to_its_consumers() -> None: + """The expansion walk counts an alias inside its own anchor once and ends. + The document still loads as the cycle it is: the contract API refuses it + with its documented ``contract_load_failed``.""" + data = load_yaml_safe("a: &a [x, *a]") + assert data["a"][1] is data["a"] + + +def test_recursive_alias_does_not_hide_an_expansion() -> None: + doc = _doubling_chain(24) + "\nloop: &loop [*l24, *loop]" + with pytest.raises(UnsafeYamlError, match="expand"): + load_yaml_safe(doc) + + +def test_budget_counts_what_aliases_add_not_document_size() -> None: + """One alias sends this 800-node document through the expansion walk; it + adds one node, so it loads under a budget of 50.""" + doc = "anchor: &a x\nref: *a\n" + "\n".join(f"k{i}: [a, b]" for i in range(200)) + assert load_yaml_safe(doc, max_alias_expansion=50)["ref"] == "x" + + +def test_ratio_cap_refuses_a_chain_under_the_node_budget() -> None: + """13 levels add 49,120 nodes, inside the 100,000 default, but grow a 30-node + document more than 1,600-fold, so the ratio rule refuses it on its own.""" + try: + load_yaml_safe(_doubling_chain(13)) + except UnsafeYamlError as refused: + assert "from 30 to 49150 nodes" in str(refused) + else: + pytest.fail("a 1,638-fold expansion loaded") + + +@pytest.mark.parametrize("refs, refused", [(39, False), (45, True)]) +def test_default_expansion_cap(refs: int, refused: bool) -> None: + """A 2,501-node block aliased 39 times adds 97,539 nodes; 45 times adds + 112,545, over the 100,000 default. Neither trips the ratio rule.""" + block = "{" + ", ".join(f"f{i}: v" for i in range(1250)) + "}" + doc = f"base: &b {block}\n" + "\n".join(f"r{i}: *b" for i in range(refs)) + if refused: + with pytest.raises(UnsafeYamlError, match="expand"): + load_yaml_safe(doc) + else: + assert len(load_yaml_safe(doc)) == refs + 1 + + +@pytest.mark.parametrize( + "tag", ["!!python/object/apply:os.getcwd []", "!!python/name:os.getcwd ''"] +) +def test_alias_path_constructs_safe_tags_only(tag: str) -> None: + """With an alias the document is composed and constructed by hand; that + path must refuse Python object tags exactly as ``yaml.safe_load`` does.""" + with pytest.raises(yaml.YAMLError): + load_yaml_safe(f"a: &x 1\nb: *x\nc: {tag}") + + +def test_stream_is_read_only_up_to_the_cap() -> None: + """A stream is refused after one unit past the cap, not read whole.""" + + class _Stream: + def __init__(self) -> None: + self.sizes: list = [] + + def read(self, size: int = -1) -> str: + self.sizes.append(size) + return "x" * (size if size > 0 else 10_000_000) + + stream = _Stream() + with pytest.raises(UnsafeYamlError, match="safety cap"): + load_yaml_safe(stream, max_bytes=1_000) + assert stream.sizes == [1_001] + + +def test_short_reads_are_read_to_the_end() -> None: + """A raw stream may return fewer units than asked before its end; the + bounded read keeps reading instead of parsing the first chunk.""" + + class _Trickle(io.RawIOBase): + def __init__(self, data: bytes) -> None: + self.data = data + + def readable(self) -> bool: + return True + + def read(self, size: int = -1) -> bytes: + chunk, self.data = self.data[:7], self.data[7:] + return chunk + + assert load_yaml_safe(_Trickle(b"name: a-long-enough-value\nid: x\n")) == { + "name": "a-long-enough-value", + "id": "x", + } + with pytest.raises(UnsafeYamlError, match="safety cap"): + load_yaml_safe(_Trickle(b"k: " + b"v" * 100), max_bytes=50) + + +@pytest.mark.parametrize("as_bytes", [False, True], ids=["str", "bytes"]) +def test_size_cap_counts_utf8_bytes(as_bytes: bool) -> None: + """400 three-byte characters are 1,203 bytes: over a 1,000-byte cap + whether they arrive as text or as bytes.""" + doc = "k: " + "\u20ac" * 400 + with pytest.raises(UnsafeYamlError, match="1000-byte"): + load_yaml_safe(doc.encode("utf-8") if as_bytes else doc, max_bytes=1_000) + assert load_yaml_safe("k: " + "\u20ac" * 300, max_bytes=1_000) == {"k": "\u20ac" * 300} + + +def test_depth_cap_boundary() -> None: + """A document nested to the cap loads; one level more is refused, not + left to PyYAML's ``RecursionError``, and so is one 20,000 levels deep, + which PyYAML's scanner took tens of seconds to read to the end. + ``a: [[1]]`` is three levels.""" + assert load_yaml_safe("[" * 128 + "]" * 128) is not None + for depth in (129, 20_000): + with pytest.raises(UnsafeYamlError, match="more than 128 levels"): + load_yaml_safe("[" * depth + "]" * depth) + assert load_yaml_safe("a: [[1]]", max_depth=3) == {"a": [[1]]} + with pytest.raises(UnsafeYamlError, match="more than 2 levels"): + load_yaml_safe("a: [[1]]", max_depth=2) + + +def test_aliases_repeating_a_long_string_are_refused() -> None: + """Twenty aliases add few nodes but repeat a 1,000,000-character string + twenty times; the node budget alone let it through.""" + big = "x" * 1_000_000 + doc = f"s: &s {big}\nl1: &l1 [*s, *s, *s, *s]\nl2: &l2 [*l1, *l1, *l1, *l1]\n" + with pytest.raises(UnsafeYamlError, match="repeat 20000000 characters"): + load_yaml_safe(doc) + + +def test_character_budget_boundary() -> None: + """Two aliases to a ten-character string add 20 characters: allowed at + 20, not at 19.""" + doc = "a: &a abcdefghij\nb: *a\nc: *a\n" + assert load_yaml_safe(doc, max_alias_expansion_chars=20)["c"] == "abcdefghij" + with pytest.raises(UnsafeYamlError, match="repeat 20 characters"): + load_yaml_safe(doc, max_alias_expansion_chars=19) + + +def test_reusing_a_block_is_allowed() -> None: + """Ordinary DRY reuse of one anchored block stays well inside the budget.""" + block = "{" + ", ".join(f"f{i}: v{i}" for i in range(100)) + "}" + refs = "\n".join(f"r{i}: *b" for i in range(40)) + data = load_yaml_safe(f"base: &b {block}\n{refs}") + assert data["r39"] == data["base"] + + +def test_expansion_budget_boundary() -> None: + """Five aliases to a ten-item list add 55 nodes: allowed at 55, not at 54.""" + doc = "base: &b [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]\n" + "\n".join(f"r{i}: *b" for i in range(5)) + assert load_yaml_safe(doc, max_alias_expansion=55)["r4"] == list(range(1, 11)) + with pytest.raises(UnsafeYamlError): + load_yaml_safe(doc, max_alias_expansion=54) + + +def test_odcs_reader_applies_the_caps(tmp_path: Path) -> None: + """ODCS/ODPS files come from other teams and catalogs; the reader the + Bitol import, the contractId resolver and ``--seed-from`` share used a + plain ``yaml.safe_load``.""" + from fluid_build.providers.odcs.io import read_input + + path = tmp_path / "bomb.odcs.yaml" + path.write_text( + "apiVersion: v3.1.0\nkind: DataContract\n" + _doubling_chain(18), encoding="utf-8" + ) + with pytest.raises(UnsafeYamlError): + read_input(path) + + +def test_callers_report_a_refused_file_as_unreadable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Callers that caught ``yaml.YAMLError`` alone crashed with + ``UnsafeYamlError``; the expansion budget refuses more inputs than the + alias count did, so each must report the file as it reports bad YAML.""" + import logging + import sys + import types + + from fluid_build.forge.core import artifact_fanout, validators + from fluid_build.forge.federation import load_federation_manifest + + bomb = _doubling_chain(13) + contract = tmp_path / "contract.fluid.yaml" + contract.write_text(bomb, encoding="utf-8") + assert artifact_fanout._contract_is_reference_only(contract) is False + assert artifact_fanout._load_schedule_contract(contract, None, logging.getLogger()) is None + # Every emitter loads through _load_contract; `fluid generate artifacts` + # reports a FanoutError and nothing else. + with pytest.raises(artifact_fanout.FanoutError, match="did not parse"): + artifact_fanout._load_contract(contract) + + manifest = tmp_path / "federation" / "upstreams.yaml" + manifest.parent.mkdir() + manifest.write_text(bomb, encoding="utf-8") + assert load_federation_manifest(tmp_path).workspaces == [] + + # The parse runs before the validator library is used. + monkeypatch.setitem(sys.modules, "openapi_spec_validator", types.ModuleType("stub")) + monkeypatch.setattr(validators, "_openapi_validator_available", lambda: True) + issues = validators.validate_openapi("api.yaml", bomb.encode(), strict=True) + assert [issue.code for issue in issues] == ["OAS-PARSE"]