Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,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.
- Preserve explicit application identities losslessly while using
collision-resistant, path-safe runtime namespace components.
- Give `BatteriesIncludedConfigLoader.cli_name` a documented identity role by
Expand Down
5 changes: 5 additions & 0 deletions docs/consumer-profiles.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
66 changes: 63 additions & 3 deletions lib/python/base_cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)):

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correctness (reproduced): _CONFIG_MAX_NODES in _validate_config_graph only increments visited_nodes for Mapping/list/tuple nodes, so a flat structure made almost entirely of scalar leaves never trips the 100,000-node limit. Reproduced: a single top-level mapping with 5,000,000 scalar leaf keys validated in ~1.8s with zero errors, even though the PR's stated goal is to reject excessive size/traversal cost. This defeats the guard for the most common shape of a bloated config.

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}."):
Expand Down Expand Up @@ -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:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Test gap: the new except RecursionError handler in load_yaml_file isn't exercised by any test — the PR's deepest test config (65 levels) is caught earlier by _validate_config_graph's own _CONFIG_MAX_DEPTH check, never reaching PyYAML's parser far enough to raise RecursionError. A future change to the depth limit or parser could silently regress this path with nothing to catch it.

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}'")

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug (message quality): load_yaml_file builds source=f"Config file '{path}'" and _validate_config_graph wraps that again as f"Configuration source '{source}' ...", producing a doubled-quote error like Configuration source 'Config file '/tmp/x.yaml'' has a non-string key... — reads like a bug in the CLI's own output rather than deliberate nested attribution.

return data
90 changes: 90 additions & 0 deletions tests/test_batteries_included_config.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

import json
import tempfile
import unittest
from pathlib import Path
Expand Down Expand Up @@ -29,6 +30,95 @@ 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)}")
self.assertEqual(result.exit_code, 2, result.output)
self.assertNotIn("RecursionError", result.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), result.output)

def test_layered_loader_merges_in_documented_order_and_records_provenance(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
Expand Down
11 changes: 11 additions & 0 deletions tests/test_optional_yaml_dependency.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,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"))
Expand Down
Loading