diff --git a/CHANGELOG.md b/CHANGELOG.md index d25dc17..42f46dd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,8 @@ and versions are tracked in the repo-root `VERSION` file. ### Fixed +- Reject missing, unreadable, and non-regular explicit `--config` paths before + profile or runtime startup while preserving optional profile-owned files. - Make `App.name` authoritative for single-command identity, reject duplicate or post-materialization registrations deterministically, stabilize inferred names across Click releases, and make module-level `@command()` functions diff --git a/README.md b/README.md index 650e8c9..d999abc 100644 --- a/README.md +++ b/README.md @@ -493,6 +493,12 @@ lifecycle after the profile's configuration is loaded; for example, `--environment prod` overrides `environment: dev` from an explicit configuration file. +An explicit `--config` value must identify an existing, readable regular file; +invalid paths are rejected as usage errors before profile loading or runtime +state begins. Invalid YAML reports that the file `contains invalid YAML`, while +a non-mapping document reports that it `must contain a YAML mapping`. Each error +includes the selected path. Quoted home-relative paths such as +`--config "~/tool.yml"` are expanded before validation. `ctx.config` exposes the dictionary returned by the profile. `ctx.user_config` exposes the opaque user-configuration value returned by the profile. Consumers diff --git a/docs/local-config.md b/docs/local-config.md index 7901956..abe63c0 100644 --- a/docs/local-config.md +++ b/docs/local-config.md @@ -5,5 +5,11 @@ Standalone applications can accept an explicit `--config` file through the generic profile, or provide their own `load_user_config` and `load_config` callbacks for application-owned configuration sources. +Command-line `--config` values are strict: the path is expanded and must be an +existing, readable regular file before profile and runtime startup. By contrast, +`base_cli.config.load_yaml_file(path)` keeps its optional-file behavior for +profile-discovered configuration; pass `required=True` when a consumer-owned +call site represents an explicit user request. + The consumer owns the configuration schema, merge semantics, and operational choice of whether to back up or synchronize its machine-local files. diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py index 5a42bc6..b761b94 100644 --- a/lib/python/base_cli/app.py +++ b/lib/python/base_cli/app.py @@ -3,6 +3,7 @@ import functools import logging import os +import stat import sys import time import traceback @@ -925,7 +926,11 @@ def decorator(func: Callable[..., Any]): def _decorate_standard_options(click: Any, func: Callable[..., Any], version: str | None): func = click.option("--log-file", type=click.Path(dir_okay=False), help="Override the persistent log file.")(func) func = click.option("--keep-temp", is_flag=True, default=None, help="Preserve this run's temp directory.")(func) - func = click.option("--config", type=click.Path(dir_okay=False), help="Load an additional config file.")(func) + func = click.option( + "--config", + type=_explicit_config_path_type(click), + help="Load an additional config file.", + )(func) func = click.option("--environment", help="Set the CLI environment.")(func) func = click.option( "--debug", @@ -945,6 +950,31 @@ def _decorate_standard_options(click: Any, func: Callable[..., Any], version: st return func +def _explicit_config_path_type(click: Any) -> Any: + class ExplicitConfigPath(click.Path): + def convert(self, value: Any, param: Any, ctx: Any) -> Path: + try: + expanded = Path(value).expanduser() + except (RuntimeError, TypeError, ValueError) as exc: + self.fail(f"Path {value!r} could not be expanded: {exc}", param, ctx) + + converted = super().convert(expanded, param, ctx) + try: + mode = converted.stat().st_mode + except OSError: + self.fail(f"Path {str(expanded)!r} does not exist.", param, ctx) + if not stat.S_ISREG(mode): + self.fail(f"Path {str(expanded)!r} is not a regular file.", param, ctx) + return converted + + return ExplicitConfigPath( + exists=True, + dir_okay=False, + readable=True, + path_type=Path, + ) + + def _pop_standard_options(kwargs: dict[str, Any]) -> dict[str, Any]: standard = {} for key in _STANDARD_OPTION_KEYS: diff --git a/lib/python/base_cli/config.py b/lib/python/base_cli/config.py index 2b51394..ab38652 100644 --- a/lib/python/base_cli/config.py +++ b/lib/python/base_cli/config.py @@ -1,5 +1,6 @@ from __future__ import annotations +import stat from pathlib import Path from typing import Any @@ -12,16 +13,37 @@ ] -def load_yaml_file(path: Path) -> dict[str, Any]: - if not path.is_file(): +def load_yaml_file(path: Path, *, required: bool = False) -> dict[str, Any]: + """Load a YAML mapping, optionally requiring a regular file to exist. + + Missing files remain an empty mapping by default so consumer profiles can + use this helper for optional implicit configuration. Callers handling an + explicitly requested file must set ``required=True``. + """ + if required: + try: + mode = path.stat().st_mode + except FileNotFoundError as exc: + raise ConfigurationError(f"Config file '{path}' does not exist.") from exc + except OSError as exc: + raise ConfigurationError(f"Unable to read config file '{path}': {exc}") from exc + if not stat.S_ISREG(mode): + raise ConfigurationError(f"Config path '{path}' is not a regular file.") + elif not path.is_file(): return {} yaml = require_yaml("PyYAML is required to load the explicit CLI configuration file.") try: contents = path.read_text(encoding="utf-8") + except FileNotFoundError as exc: + if required: + raise ConfigurationError(f"Config file '{path}' does not exist.") from exc + return {} except OSError as exc: raise ConfigurationError(f"Unable to read config file '{path}': {exc}") from exc + except UnicodeDecodeError as exc: + raise ConfigurationError(f"Unable to read config file '{path}': {exc}") from exc try: data = yaml.safe_load(contents) except yaml.YAMLError as exc: diff --git a/lib/python/base_cli/profile.py b/lib/python/base_cli/profile.py index 427e7a0..6349637 100644 --- a/lib/python/base_cli/profile.py +++ b/lib/python/base_cli/profile.py @@ -114,7 +114,7 @@ def _empty_user_config() -> None: def _load_explicit_config(_project: ProjectInfo | None, explicit: Path | None) -> dict[str, Any]: - return load_yaml_file(explicit) if explicit is not None else {} + return load_yaml_file(explicit, required=True) if explicit is not None else {} def _generic_runtime_resolver( diff --git a/tests/test_explicit_config_validation.py b/tests/test_explicit_config_validation.py new file mode 100644 index 0000000..a2cc054 --- /dev/null +++ b/tests/test_explicit_config_validation.py @@ -0,0 +1,324 @@ +from __future__ import annotations + +import importlib.util +import os +import tempfile +import unittest +from dataclasses import replace +from pathlib import Path +from typing import Any +from unittest import mock + +import base_cli +import base_cli.app as app_module +from base_cli._lifecycle import RunRecorder +from base_cli.config import load_yaml_file +from base_cli.testing import invoke + + +def _combined_output(result: Any) -> str: + output = result.output + try: + stderr = result.stderr + except ValueError: + stderr = "" + return output if not stderr or stderr in output else output + stderr + + +@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") +class ExplicitConfigValidationTests(unittest.TestCase): + def _app_with_forbidden_profile_loaders( + self, + calls: list[str], + ) -> base_cli.App: + def forbidden(name: str): + def callback(*_args: object) -> object: + calls.append(name) + raise AssertionError(f"{name} must not run before explicit-path validation") + + return callback + + profile = base_cli.CliProfile.generic( + discover_project=forbidden("discover_project"), + load_user_config=forbidden("load_user_config"), + load_config=forbidden("load_config"), + resolve_workspace_root=forbidden("resolve_workspace_root"), + ) + profile = replace( + profile, + resolve_runtime=forbidden("resolve_runtime"), + history_writer=forbidden("history_writer"), + ) + return base_cli.App(name="strict-config-path", profile=profile) + + def _assert_no_runtime_artifacts(self, home: Path) -> None: + cache_root = home / ".cache" + self.assertFalse(cache_root.exists()) + self.assertEqual(list(cache_root.rglob("run.json")), []) + self.assertEqual(list(cache_root.rglob("*.log")), []) + + def test_missing_explicit_config_is_rejected_before_profile_or_runtime_startup(self) -> None: + calls: list[str] = [] + command_calls: list[None] = [] + app = self._app_with_forbidden_profile_loaders(calls) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + command_calls.append(None) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + missing = home / "missing-config.yml" + with ( + mock.patch.object(app_module, "configure_logger") as configure_logger, + mock.patch.object(RunRecorder, "start") as start_metadata, + ): + result = invoke(app, ["--config", str(missing)], home=home) + + output = _combined_output(result) + self.assertEqual(result.exit_code, 2, output) + self.assertIn(str(missing), output) + self.assertIn("does not exist", output) + self.assertNotIn("Traceback", output) + self.assertEqual(calls, []) + self.assertEqual(command_calls, []) + configure_logger.assert_not_called() + start_metadata.assert_not_called() + self._assert_no_runtime_artifacts(home) + + def test_unreadable_explicit_config_is_rejected_before_profile_or_runtime_startup(self) -> None: + import click.types + + calls: list[str] = [] + command_calls: list[None] = [] + app = self._app_with_forbidden_profile_loaders(calls) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + command_calls.append(None) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + config = home / "unreadable-config.yml" + config.write_text("environment: test\n", encoding="utf-8") + real_access = os.access + + def deny_config_read(path: os.PathLike[str] | str, mode: int) -> bool: + if Path(path) == config and mode == os.R_OK: + return False + return real_access(path, mode) + + with ( + mock.patch.object(click.types.os, "access", side_effect=deny_config_read), + mock.patch.object(app_module, "configure_logger") as configure_logger, + mock.patch.object(RunRecorder, "start") as start_metadata, + ): + result = invoke(app, ["--config", str(config)], home=home) + + output = _combined_output(result) + self.assertEqual(result.exit_code, 2, output) + self.assertIn(str(config), output) + self.assertIn("is not readable", output) + self.assertNotIn("Traceback", output) + self.assertEqual(calls, []) + self.assertEqual(command_calls, []) + configure_logger.assert_not_called() + start_metadata.assert_not_called() + self._assert_no_runtime_artifacts(home) + + @unittest.skipUnless(hasattr(os, "mkfifo"), "FIFO creation is not available") + def test_non_regular_explicit_config_is_rejected_before_profile_or_runtime_startup(self) -> None: + calls: list[str] = [] + command_calls: list[None] = [] + app = self._app_with_forbidden_profile_loaders(calls) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + command_calls.append(None) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + config = home / "config.fifo" + os.mkfifo(config) + with ( + mock.patch.object(app_module, "configure_logger") as configure_logger, + mock.patch.object(RunRecorder, "start") as start_metadata, + ): + result = invoke(app, ["--config", str(config)], home=home) + + output = _combined_output(result) + self.assertEqual(result.exit_code, 2, output) + self.assertIn(str(config), output) + self.assertIn("is not a regular file", output) + self.assertEqual(calls, []) + self.assertEqual(command_calls, []) + configure_logger.assert_not_called() + start_metadata.assert_not_called() + self._assert_no_runtime_artifacts(home) + + def test_explicit_config_removed_after_parsing_is_not_silently_ignored(self) -> None: + discovery_calls: list[None] = [] + lifecycle_calls: list[str] = [] + command_calls: list[None] = [] + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + config = home / "disappearing-config.yml" + config.write_text("environment: test\n", encoding="utf-8") + + def remove_config(_cwd: Path) -> None: + discovery_calls.append(None) + config.unlink() + + def forbidden(name: str): + def callback(*_args: object) -> object: + lifecycle_calls.append(name) + raise AssertionError(f"{name} must not run after explicit config disappears") + + return callback + + profile = replace( + base_cli.CliProfile.generic(discover_project=remove_config), + resolve_runtime=forbidden("resolve_runtime"), + history_writer=forbidden("history_writer"), + ) + app = base_cli.App(name="disappearing-config", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + command_calls.append(None) + + with ( + mock.patch.object(app_module, "configure_logger") as configure_logger, + mock.patch.object(RunRecorder, "start") as start_metadata, + ): + result = invoke(app, ["--config", str(config)], home=home) + + output = _combined_output(result) + self.assertEqual(result.exit_code, 2, output) + self.assertIn(f"Config file '{config}' does not exist.", output) + self.assertEqual(discovery_calls, [None]) + self.assertEqual(lifecycle_calls, []) + self.assertEqual(command_calls, []) + configure_logger.assert_not_called() + start_metadata.assert_not_called() + self._assert_no_runtime_artifacts(home) + + def test_malformed_explicit_yaml_is_rejected_before_runtime_side_effects(self) -> None: + self._assert_config_content_rejection( + contents="broken: [\n", + expected_message="Config file '{path}' contains invalid YAML:", + ) + + def test_non_mapping_explicit_yaml_is_rejected_before_runtime_side_effects(self) -> None: + self._assert_config_content_rejection( + contents="- first\n- second\n", + expected_message="Config file '{path}' must contain a YAML mapping.", + ) + + def test_invalid_utf8_explicit_config_is_rejected_before_runtime_side_effects(self) -> None: + self._assert_config_content_rejection( + contents=b"environment: \xff\xfe\n", + expected_message="Unable to read config file '{path}':", + ) + + def _assert_config_content_rejection( + self, + *, + contents: str | bytes, + expected_message: str, + ) -> None: + lifecycle_calls: list[str] = [] + command_calls: list[None] = [] + + def forbidden(name: str): + def callback(*_args: object) -> object: + lifecycle_calls.append(name) + raise AssertionError(f"{name} must not run after invalid explicit config") + + return callback + + profile = replace( + base_cli.CliProfile.generic(), + resolve_runtime=forbidden("resolve_runtime"), + history_writer=forbidden("history_writer"), + ) + app = base_cli.App(name="invalid-config-content", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + command_calls.append(None) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + config = home / "invalid-config.yml" + if isinstance(contents, bytes): + config.write_bytes(contents) + else: + config.write_text(contents, encoding="utf-8") + with ( + mock.patch.object(app_module, "configure_logger") as configure_logger, + mock.patch.object(RunRecorder, "start") as start_metadata, + ): + result = invoke(app, ["--config", str(config)], home=home) + + output = _combined_output(result) + self.assertEqual(result.exit_code, 2, output) + self.assertIn(expected_message.format(path=config), output) + self.assertNotIn("Traceback", output) + self.assertEqual(lifecycle_calls, []) + self.assertEqual(command_calls, []) + configure_logger.assert_not_called() + start_metadata.assert_not_called() + self._assert_no_runtime_artifacts(home) + + def test_absent_implicit_profile_config_remains_optional(self) -> None: + seen: dict[str, object] = {} + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + implicit = home / "optional-profile-config.yml" + profile = base_cli.CliProfile.generic( + load_user_config=lambda: load_yaml_file(implicit), + ) + app = base_cli.App( + name="optional-implicit-config", + profile=profile, + log_to_file=False, + ) + + @app.command() + def main(ctx: base_cli.Context) -> None: + seen["user_config"] = ctx.user_config + + result = invoke(app, home=home) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(seen, {"user_config": {}}) + self.assertFalse(implicit.exists()) + + def test_explicit_config_validation_preserves_tilde_expansion(self) -> None: + seen: dict[str, object] = {} + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + config = home / "explicit-config.yml" + config.write_text("environment: tilde\n", encoding="utf-8") + app = base_cli.App(name="tilde-config", log_to_file=False) + + @app.command() + def main(ctx: base_cli.Context) -> None: + seen["environment"] = ctx.environment + + result = invoke(app, ["--config", "~/explicit-config.yml"], home=home) + + self.assertEqual(result.exit_code, 0, _combined_output(result)) + self.assertEqual(seen, {"environment": "tilde"}) + + +if __name__ == "__main__": + unittest.main()