fix: validate nested configuration value graphs

CHANGELOG.md

@@ -23,6 +23,9 @@ and versions are tracked in the repo-root `VERSION` file.
 
 ### Fixed
 
+- Validate nested configuration mappings before merge/provenance traversal,
+  reject recursive or excessively deep values with source-aware errors, and
+  continue to accept shared YAML aliases.
 - Reject recursive and concurrent in-process `run_app()` calls before they can
   replace another invocation's stdout or logging handlers.
 - Rotate bounded byte-retention size walks across invocations with a persisted

docs/consumer-profiles.md

@@ -144,6 +144,11 @@ explicit, project, or user base `environment` value is used, falling back to
 lower-precedence value. `Context.config_provenance` records the winning source
 for each dotted key.
 
+All mapping keys must be strings, including keys nested inside sequences.
+Recursive configuration values are rejected, nesting is limited to 64 levels,
+and shared YAML aliases are accepted when they do not form a cycle. Errors name
+the configuration source and relevant nested path.
+
 The reserved framework keys `environment`, `log_level`, and `keep_temp` are
 validated into `Context.framework_config` and are excluded from the consumer
 configuration dictionary. All other keys remain consumer-owned and are exposed

lib/python/base_cli/config.py

@@ -26,6 +26,8 @@
 _LOG_LEVELS = frozenset({"debug", "info", "warning", "error", "critical"})
 _SAFE_NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*\Z")
 _SAFE_FILENAME = re.compile(r"(?:[A-Za-z0-9][A-Za-z0-9_.-]*|\.[A-Za-z0-9][A-Za-z0-9_.-]*)\Z")
+_CONFIG_MAX_DEPTH = 64
+_CONFIG_MAX_NODES = 100_000
 
 
 @dataclass(frozen=True)
@@ -100,21 +102,74 @@ def _leaf_provenance(
     return {prefix: source} if prefix else {}
 
 
+def _validate_config_graph(value: Mapping[str, Any], *, source: str) -> None:
+    """Validate nested mapping keys, cycles, depth, and traversal cost."""
+
+    active: set[int] = set()
+    stack: list[tuple[bool, Any, str, int]] = [(False, value, "", 0)]
+    visited_nodes = 0
+    while stack:
+        exiting, current, path, depth = stack.pop()
+        identity = id(current)
+        if exiting:
+            active.remove(identity)
+            continue
+        visited_nodes += 1
+        if visited_nodes > _CONFIG_MAX_NODES:
+            raise ConfigurationError(
+                f"Configuration source {source} exceeds the maximum of {_CONFIG_MAX_NODES} nested values."
+            )
+        if not isinstance(current, (Mapping, list, tuple)):
+            continue
+        if identity in active:
+            location = path or "<root>"
+            raise ConfigurationError(f"Configuration source {source} contains a recursive value at '{location}'.")
+        if depth > _CONFIG_MAX_DEPTH:
+            location = path or "<root>"
+            raise ConfigurationError(
+                f"Configuration source {source} exceeds the maximum nesting depth of {_CONFIG_MAX_DEPTH} at "
+                f"'{location}'."
+            )
+        active.add(identity)
+        stack.append((True, current, path, depth))
+        if isinstance(current, Mapping):
+            children = list(current.items())
+            for key, child in reversed(children):
+                if not isinstance(key, str):
+                    location = path or "<root>"
+                    raise ConfigurationError(f"Configuration source {source} has a non-string key under '{location}'.")
+                child_path = f"{path}.{key}" if path else key
+                stack.append((False, child, child_path, depth + 1))
+        else:
+            for index, child in reversed(tuple(enumerate(current))):
+                stack.append((False, child, f"{path}[{index}]", depth + 1))
+
+
 def _merge_mapping(
     target: dict[str, Any],
     provenance: dict[str, str],
     incoming: Mapping[str, Any],
     source: str,
     *,
     prefix: str = "",
+) -> None:
+    _validate_config_graph(incoming, source=source)
+    _merge_mapping_validated(target, provenance, incoming, source, prefix=prefix)
+
+
+def _merge_mapping_validated(
+    target: dict[str, Any],
+    provenance: dict[str, str],
+    incoming: Mapping[str, Any],
+    source: str,
+    *,
+    prefix: str = "",
 ) -> None:
     for key, value in incoming.items():
-        if not isinstance(key, str):
-            raise ConfigurationError("Configuration keys must be strings.")
         path = f"{prefix}.{key}" if prefix else key
         previous = target.get(key)
         if isinstance(previous, Mapping) and isinstance(value, Mapping):
-            _merge_mapping(target[key], provenance, value, source, prefix=path)
+            _merge_mapping_validated(target[key], provenance, value, source, prefix=path)
             continue
         for existing_path in tuple(provenance):
             if existing_path == path or existing_path.startswith(f"{path}."):
@@ -267,10 +322,15 @@ def load_yaml_file(path: Path, *, required: bool = False) -> dict[str, Any]:
         raise ConfigurationError(f"Unable to read config file '{path}': {exc}") from exc
     try:
         data = yaml.safe_load(contents)
+    except RecursionError as exc:
+        raise ConfigurationError(
+            f"Config file '{path}' exceeds the maximum nesting depth of {_CONFIG_MAX_DEPTH}."
+        ) from exc
     except yaml.YAMLError as exc:
         raise ConfigurationError(f"Config file '{path}' contains invalid YAML: {exc}") from exc
     if data is None:
         return {}
     if not isinstance(data, dict):
         raise ConfigurationError(f"Config file '{path}' must contain a YAML mapping.")
+    _validate_config_graph(data, source=f"Config file '{path}'")
     return data

tests/test_batteries_included_config.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import json
 import tempfile
 import unittest
 from pathlib import Path
+from typing import Any
 from unittest.mock import patch
 
 import base_cli
@@ -15,6 +17,17 @@ def _write_yaml(path: Path, contents: str) -> None:
     path.write_text(contents, encoding="utf-8")
 
 
+def _combined_output(result: Any) -> str:
+    """Return human-readable output across Click's split/combined result APIs."""
+
+    output = result.output
+    try:
+        stderr = result.stderr
+    except ValueError:
+        stderr = ""
+    return output if not stderr or stderr in output else output + stderr
+
+
 class BatteriesIncludedConfigTests(unittest.TestCase):
     def test_nested_merge_provenance_keeps_repeated_leaf_names_scoped(self) -> None:
         values: dict[str, object] = {}
@@ -29,6 +42,96 @@ def test_nested_merge_provenance_keeps_repeated_leaf_names_scoped(self) -> None:
             {"host": "user", "db.host": "project", "db.tls.enabled": "project"},
         )
 
+    def test_nested_non_string_keys_are_rejected_before_first_insert_or_overlay(self) -> None:
+        for initial in ({}, {"nested": {"keep": True}}):
+            with self.subTest(initial=initial):
+                values = dict(initial)
+                provenance: dict[str, str] = {"existing": "prior"}
+                original_values = dict(values)
+                original_provenance = dict(provenance)
+                with self.assertRaisesRegex(base_cli.ConfigurationError, "nested"):
+                    _merge_mapping(values, provenance, {"nested": {2: "invalid"}}, "explicit")
+                self.assertEqual(values, original_values)
+                self.assertEqual(provenance, original_provenance)
+
+    def test_shared_mapping_alias_is_valid_but_recursive_alias_is_rejected_with_path(self) -> None:
+        shared = {"answer": 42}
+        valid = {"left": shared, "right": shared}
+        values: dict[str, object] = {}
+        provenance: dict[str, str] = {}
+        _merge_mapping(values, provenance, valid, "user")
+        self.assertEqual(values, {"left": shared, "right": shared})
+        self.assertEqual(provenance, {"left.answer": "user", "right.answer": "user"})
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "recursive.yaml"
+            _write_yaml(path, "nested: &node\n  child: *node\n")
+            with self.assertRaisesRegex(base_cli.ConfigurationError, "recursive.yaml.*recursive value.*nested.child"):
+                BatteriesIncludedConfigLoader(user_config_dir=Path(tmpdir) / "user").load(None, path)
+
+    def test_mapping_depth_is_bounded_before_recursive_merge_or_provenance(self) -> None:
+        nested: dict[str, object] = {"value": 1}
+        for index in range(65):
+            nested = {f"level{index}": nested}
+
+        with self.assertRaisesRegex(base_cli.ConfigurationError, "maximum nesting depth of 64"):
+            _merge_mapping({}, {}, nested, "explicit")
+
+    def test_scalar_nodes_count_toward_configuration_graph_limit(self) -> None:
+        with self.assertRaisesRegex(base_cli.ConfigurationError, "maximum of 100000 nested values"):
+            _merge_mapping({}, {}, {"values": [0] * 100_000}, "explicit")
+
+    def test_nested_invalid_yaml_shape_is_usage_error_in_human_and_json_modes(self) -> None:
+        import click
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            config_path = root / "invalid.yaml"
+            _write_yaml(config_path, "nested:\n  2: invalid\n")
+            profile = base_cli.CliProfile.batteries_included(
+                "invalid-nested-config",
+                user_config_dir=root / "user",
+            )
+            app = base_cli.App(
+                name="invalid-nested-config",
+                profile=profile,
+                log_to_file=False,
+                lifecycle_options=base_cli.LifecycleOptions(json=base_cli.LifecycleOption("--json")),
+            )
+
+            @app.command()
+            def main(ctx: base_cli.Context) -> None:
+                del ctx
+
+            @click.command(name="invalid-nested-config")
+            def attached_command() -> None:
+                pass
+
+            attached_app = base_cli.App(
+                name="invalid-nested-config",
+                profile=base_cli.CliProfile.batteries_included(
+                    "invalid-nested-config",
+                    user_config_dir=root / "user",
+                ),
+                log_to_file=False,
+                lifecycle_options=base_cli.LifecycleOptions(json=base_cli.LifecycleOption("--json")),
+            )
+            attached = attached_app.attach(attached_command)
+            targets = (("native", app), ("attached", attached))
+            for target_name, target in targets:
+                for args in (["--config", str(config_path)], ["--json", "--config", str(config_path)]):
+                    with self.subTest(target=target_name, json="--json" in args):
+                        result = invoke(target, list(args), home=root / f"home-{target_name}-{len(args)}")
+                        output = _combined_output(result)
+                        self.assertEqual(result.exit_code, 2, output)
+                        self.assertNotIn("RecursionError", output)
+                        if "--json" in args:
+                            payload = json.loads(result.stdout)
+                            self.assertEqual(payload["code"], "usage_error")
+                            self.assertIn(str(config_path), payload["message"])
+                        else:
+                            self.assertIn(str(config_path), output)
+
     def test_layered_loader_merges_in_documented_order_and_records_provenance(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             root = Path(tmpdir)

tests/test_optional_yaml_dependency.py

@@ -67,6 +67,17 @@ def test_yaml_config_explains_optional_install_when_yaml_is_missing(self) -> Non
                 with self.assertRaisesRegex(ConfigurationError, r"base-cli\[yaml\]"):
                     load_yaml_file(path, required=True)
 
+    def test_yaml_config_converts_parser_recursion_error(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "config.yaml"
+            path.write_text("answer: 42\n", encoding="utf-8")
+            yaml = mock.Mock()
+            yaml.safe_load.side_effect = RecursionError("parser recursion")
+            yaml.YAMLError = type("YAMLError", (Exception,), {})
+            with mock.patch("base_cli.config.require_yaml", return_value=yaml):
+                with self.assertRaisesRegex(ConfigurationError, r"maximum nesting depth of 64"):
+                    load_yaml_file(path, required=True)
+
     def test_core_facade_import_does_not_import_yaml(self) -> None:
         self.assertIn("base_cli", sys.modules)
         self.assertTrue(hasattr(base_cli, "App"))