fix(cli): reject corrupt config in noninteractive runs
This commit is contained in:
embwl0x
@@ -45,6 +45,10 @@ logger = logging.getLogger(__name__)
|
||||
_CONFIG_PARSE_WARNED: set = set()
|
||||
|
||||
|
||||
class InvalidUserConfigError(RuntimeError):
|
||||
"""Raised when a run that cannot repair config finds invalid user YAML."""
|
||||
|
||||
|
||||
def _backup_corrupt_config(config_path: Path) -> Optional[Path]:
|
||||
"""Preserve a corrupted ``config.yaml`` by copying it to a timestamped ``.bak``.
|
||||
|
||||
@@ -771,6 +775,48 @@ def get_config_path() -> Path:
|
||||
"""Get the main config file path."""
|
||||
return get_hermes_home() / "config.yaml"
|
||||
|
||||
|
||||
def require_parseable_user_config(*, ignore_user_config: bool = False) -> None:
|
||||
"""Reject an existing invalid config before a non-interactive agent run.
|
||||
|
||||
Interactive surfaces retain ``load_config()``'s recovery behavior so the
|
||||
operator can repair their configuration. A one-shot or single-query run
|
||||
has no such repair opportunity: allowing defaults there can silently pick
|
||||
a hosted provider/model and spend against credentials loaded from ``.env``.
|
||||
|
||||
Missing and empty files remain valid first-run states. The explicit
|
||||
``--ignore-user-config``/safe-mode escape hatch also remains authoritative.
|
||||
"""
|
||||
if ignore_user_config or os.environ.get("HERMES_IGNORE_USER_CONFIG") == "1":
|
||||
return
|
||||
|
||||
config_path = get_config_path()
|
||||
try:
|
||||
with open(config_path, encoding="utf-8") as f:
|
||||
data = fast_safe_load(f)
|
||||
except FileNotFoundError:
|
||||
return
|
||||
except Exception as exc:
|
||||
parse_error = exc
|
||||
else:
|
||||
if data is None or isinstance(data, dict):
|
||||
return
|
||||
parse_error = TypeError(
|
||||
f"top-level YAML value must be a mapping, got {type(data).__name__}"
|
||||
)
|
||||
|
||||
backup_path = _backup_corrupt_config(config_path)
|
||||
message = (
|
||||
f"Refusing non-interactive startup because {config_path} is invalid: "
|
||||
f"{parse_error}. Repair the file or pass --ignore-user-config to "
|
||||
"intentionally run with built-in defaults."
|
||||
)
|
||||
if backup_path is not None:
|
||||
message += f" A copy was saved to {backup_path}."
|
||||
logger.error(message)
|
||||
raise InvalidUserConfigError(message) from parse_error
|
||||
|
||||
|
||||
def get_env_path() -> Path:
|
||||
"""Get the .env file path (for API keys)."""
|
||||
return get_hermes_home() / ".env"
|
||||
|
||||
+43
-10
@@ -3190,9 +3190,10 @@ def _resolve_use_tui(args) -> bool:
|
||||
|
||||
def cmd_chat(args):
|
||||
"""Run interactive chat CLI."""
|
||||
use_tui = _resolve_use_tui(args)
|
||||
|
||||
_apply_safe_mode(args)
|
||||
_apply_user_config_bypass(args)
|
||||
_guard_noninteractive_user_config(args)
|
||||
use_tui = _resolve_use_tui(args)
|
||||
|
||||
# --in DIR: run in DIR. Must happen before any session resolution so the
|
||||
# workspace-scoped "latest"/-c lookups key off DIR, and it pins the
|
||||
@@ -3425,14 +3426,6 @@ def cmd_chat(args):
|
||||
if getattr(args, "yolo", False):
|
||||
os.environ["HERMES_YOLO_MODE"] = "1"
|
||||
|
||||
# --ignore-user-config: make load_cli_config() / load_config() skip the
|
||||
# user's ~/.hermes/config.yaml and return built-in defaults. Set BEFORE
|
||||
# importing cli (which runs `CLI_CONFIG = load_cli_config()` at module
|
||||
# import time). Credentials in .env are still loaded — this flag only
|
||||
# ignores behavioral/config settings.
|
||||
if getattr(args, "ignore_user_config", False):
|
||||
os.environ["HERMES_IGNORE_USER_CONFIG"] = "1"
|
||||
|
||||
# --ignore-rules: skip auto-injection of AGENTS.md/SOUL.md/.cursorrules
|
||||
# (rules), memory entries, and any preloaded skills coming from user config.
|
||||
# Maps to AIAgent(skip_context_files=True, skip_memory=True).
|
||||
@@ -12644,6 +12637,8 @@ def _prepare_agent_startup(args) -> None:
|
||||
if getattr(args, "yolo", False):
|
||||
os.environ["HERMES_YOLO_MODE"] = "1"
|
||||
_apply_safe_mode(args)
|
||||
_apply_user_config_bypass(args)
|
||||
_guard_noninteractive_user_config(args)
|
||||
|
||||
_sub_attr, _sub_set = _AGENT_SUBCOMMANDS.get(args.command, (None, None))
|
||||
if not (
|
||||
@@ -12735,6 +12730,44 @@ def _apply_safe_mode(args) -> None:
|
||||
os.environ["HERMES_IGNORE_RULES"] = "1"
|
||||
|
||||
|
||||
def _apply_user_config_bypass(args) -> None:
|
||||
"""Apply the explicit config bypass before any startup config reads."""
|
||||
if getattr(args, "ignore_user_config", False):
|
||||
os.environ["HERMES_IGNORE_USER_CONFIG"] = "1"
|
||||
|
||||
|
||||
def _guard_noninteractive_user_config(args) -> None:
|
||||
"""Fail closed before a non-interactive invocation initializes providers."""
|
||||
if getattr(args, "_noninteractive_config_validated", False):
|
||||
return
|
||||
|
||||
is_noninteractive = (
|
||||
bool(getattr(args, "oneshot", None))
|
||||
or getattr(args, "query", None) is not None
|
||||
or bool(getattr(args, "quiet", False))
|
||||
)
|
||||
if not is_noninteractive:
|
||||
return
|
||||
|
||||
from hermes_cli.config import (
|
||||
InvalidUserConfigError,
|
||||
require_parseable_user_config,
|
||||
)
|
||||
|
||||
try:
|
||||
require_parseable_user_config(
|
||||
ignore_user_config=bool(
|
||||
getattr(args, "ignore_user_config", False)
|
||||
or getattr(args, "safe_mode", False)
|
||||
)
|
||||
)
|
||||
except InvalidUserConfigError as exc:
|
||||
print(f"Error: {exc}", file=sys.stderr)
|
||||
raise SystemExit(2) from exc
|
||||
|
||||
setattr(args, "_noninteractive_config_validated", True)
|
||||
|
||||
|
||||
def _set_chat_arg_defaults(args) -> None:
|
||||
for attr, default in [
|
||||
("query", None),
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
"""Regression coverage for fail-closed non-interactive config startup."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
from argparse import Namespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _args(**overrides) -> Namespace:
|
||||
values = {
|
||||
"command": "chat",
|
||||
"ignore_user_config": False,
|
||||
"oneshot": None,
|
||||
"query": "hello",
|
||||
"quiet": False,
|
||||
"safe_mode": False,
|
||||
"yolo": False,
|
||||
}
|
||||
values.update(overrides)
|
||||
return Namespace(**values)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolated_config_env(monkeypatch, tmp_path):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
monkeypatch.delenv("HERMES_IGNORE_USER_CONFIG", raising=False)
|
||||
yield
|
||||
os.environ.pop("HERMES_IGNORE_USER_CONFIG", None)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"args",
|
||||
[
|
||||
_args(query="hello"),
|
||||
_args(command=None, query=None, oneshot="hello"),
|
||||
_args(query=None, quiet=True),
|
||||
],
|
||||
ids=["single-query", "oneshot", "quiet"],
|
||||
)
|
||||
def test_noninteractive_guard_rejects_malformed_yaml(args, tmp_path, caplog, capsys):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
broken = "model: [unterminated\n"
|
||||
config_path = tmp_path / "config.yaml"
|
||||
config_path.write_text(broken, encoding="utf-8")
|
||||
|
||||
with caplog.at_level(logging.ERROR, logger="hermes_cli.config"):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
main_mod._guard_noninteractive_user_config(args)
|
||||
|
||||
assert exc_info.value.code == 2
|
||||
assert "Refusing non-interactive startup" in capsys.readouterr().err
|
||||
assert any(
|
||||
record.levelno == logging.ERROR
|
||||
and "Refusing non-interactive startup" in record.getMessage()
|
||||
for record in caplog.records
|
||||
)
|
||||
assert config_path.read_text(encoding="utf-8") == broken
|
||||
backups = list(tmp_path.glob("config.yaml.corrupt.*.bak"))
|
||||
assert len(backups) == 1
|
||||
assert backups[0].read_text(encoding="utf-8") == broken
|
||||
|
||||
|
||||
def test_prepare_rejects_bad_config_before_plugin_discovery(monkeypatch, tmp_path):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
(tmp_path / "config.yaml").write_text("model: [unterminated\n")
|
||||
discovery_calls = []
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"hermes_cli.plugins",
|
||||
types.SimpleNamespace(
|
||||
discover_plugins=lambda: discovery_calls.append("plugins")
|
||||
),
|
||||
)
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
main_mod._prepare_agent_startup(_args())
|
||||
|
||||
assert exc_info.value.code == 2
|
||||
assert discovery_calls == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content", [None, "", "{}\n", "model:\n default: local/test\n"]
|
||||
)
|
||||
def test_noninteractive_guard_accepts_missing_empty_and_mapping_configs(
|
||||
content, tmp_path
|
||||
):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
if content is not None:
|
||||
(tmp_path / "config.yaml").write_text(content, encoding="utf-8")
|
||||
args = _args()
|
||||
|
||||
main_mod._guard_noninteractive_user_config(args)
|
||||
|
||||
assert args._noninteractive_config_validated is True
|
||||
|
||||
|
||||
def test_noninteractive_guard_rejects_non_mapping_yaml(tmp_path, capsys):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
(tmp_path / "config.yaml").write_text("- model\n- provider\n")
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
main_mod._guard_noninteractive_user_config(_args())
|
||||
|
||||
assert exc_info.value.code == 2
|
||||
assert "top-level YAML value must be a mapping" in capsys.readouterr().err
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"args",
|
||||
[
|
||||
_args(ignore_user_config=True),
|
||||
_args(safe_mode=True),
|
||||
],
|
||||
ids=["ignore-user-config", "safe-mode"],
|
||||
)
|
||||
def test_explicit_config_bypasses_allow_noninteractive_recovery(args, tmp_path):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
(tmp_path / "config.yaml").write_text("model: [unterminated\n")
|
||||
|
||||
main_mod._guard_noninteractive_user_config(args)
|
||||
|
||||
assert args._noninteractive_config_validated is True
|
||||
assert list(tmp_path.glob("config.yaml.corrupt.*.bak")) == []
|
||||
|
||||
|
||||
def test_interactive_chat_keeps_existing_repair_behavior(tmp_path):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
(tmp_path / "config.yaml").write_text("model: [unterminated\n")
|
||||
args = _args(query=None)
|
||||
|
||||
main_mod._guard_noninteractive_user_config(args)
|
||||
|
||||
assert not hasattr(args, "_noninteractive_config_validated")
|
||||
assert list(tmp_path.glob("config.yaml.corrupt.*.bak")) == []
|
||||
|
||||
|
||||
def test_ignore_user_config_is_applied_before_oneshot_startup(monkeypatch):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
monkeypatch.delenv("HERMES_IGNORE_USER_CONFIG", raising=False)
|
||||
|
||||
main_mod._apply_user_config_bypass(_args(ignore_user_config=True))
|
||||
|
||||
assert os.environ["HERMES_IGNORE_USER_CONFIG"] == "1"
|
||||
Reference in New Issue
Block a user