Skip to content
Closed
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 hermes_cli/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
get_config_path,
read_raw_config,
require_readable_config_before_write,
_refuse_write_if_unparsable,
)
from hermes_constants import OPENROUTER_BASE_URL, secure_parent_dir
from agent.credential_persistence import sanitize_borrowed_credential_payload
Expand Down Expand Up @@ -6660,6 +6661,8 @@ def _update_config_for_provider(
config_path = get_config_path()
config_path.parent.mkdir(parents=True, exist_ok=True)
require_readable_config_before_write(config_path)
if _refuse_write_if_unparsable(config_path, label="set provider config"):
return config_path

config = read_raw_config()

Expand Down
55 changes: 55 additions & 0 deletions hermes_cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -7014,6 +7014,51 @@ def require_readable_config_before_write(config_path: Optional[Path] = None) ->
) from exc


def _refuse_write_if_unparsable(config_path: Path, label: str = "write config") -> bool:
"""Return True if a config write must be refused because the existing file is
present, non-empty, and either fails to parse OR is not a YAML mapping.

Both conditions would let ``read_raw_config()`` silently collapse the file
to ``{}`` β€” a parse exception returns ``{}`` (line ~6982), and a ``[]``/scalar
root is coerced to ``{}`` at the same spot β€” causing ``save_config`` /
``set_config_value`` to wipe every non-default section on the next write. A
genuinely valid empty ``{}`` config parses cleanly and is allowed through.

*label* customises the refusal log line so each caller (save_config,
set_config_value, the auth provider writer) can be matched by its own test.

This is the single shared chokepoint for the parse-guard bug class: used by
``save_config``, ``set_config_value``, and the auth provider writer so the
guard cannot be skipped by routing the write through a different entry point.
"""
try:
_existing_size = config_path.stat().st_size
except OSError:
_existing_size = 0
if _existing_size == 0:
return False # new or empty file β€” nothing to clobber; allow write
try:
with open(config_path, encoding="utf-8") as _guard_f:
_parsed = fast_safe_load(_guard_f)
except Exception as _exc:
logger.warning(
"Refusing to %s: %s is non-empty (%d bytes) but failed to "
"parse (%s). Writing now would wipe all non-default sections. Fix the "
"YAML or retry.",
label, config_path, _existing_size, _exc,
)
return True
if not isinstance(_parsed, dict):
logger.warning(
"Refusing to %s: %s is non-empty (%d bytes) but its root is a "
"%s, not a mapping. Writing now would overwrite it with a mapping and "
"lose the existing content. Fix the YAML or retry.",
label, config_path, _existing_size, type(_parsed).__name__,
)
return True
return False


def atomic_config_write(config_path: Path, data: Any, **kwargs: Any) -> None:
"""Fail-closed atomic write for ``config.yaml``.

Expand Down Expand Up @@ -7465,6 +7510,13 @@ def save_config(
ensure_hermes_home()
config_path = get_config_path()
require_readable_config_before_write(config_path)
# ---- Parse guard: refuse write if existing file is unparseable ----
# Shared chokepoint for the config-write safety class. A valid empty {}
# config passes through; a parse failure OR a non-mapping root (list/
# scalar) is refused because read_raw_config() would otherwise collapse
# it to {} and wipe every non-default section.
if _refuse_write_if_unparsable(config_path, label="save config"):
return
# Compute explicit user paths BEFORE any normalisation --------
# _normalize_max_turns_config may inject agent.max_turns from
# DEFAULT_CONFIG; using the raw dict preserves which paths the
Expand Down Expand Up @@ -8454,6 +8506,9 @@ def set_config_value(key: str, value: str):
# dumping all default values back to the file
config_path = get_config_path()
require_readable_config_before_write(config_path)
# ---- Parse guard (shared chokepoint, same as save_config) ----
if _refuse_write_if_unparsable(config_path, label="set config value"):
return
user_config = {}
if config_path.exists():
try:
Expand Down
272 changes: 272 additions & 0 deletions tests/hermes_cli/test_save_config_wipe_guard.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,272 @@
"""Tests for save_config/set_config_value parse-failure guard.

Verifies the inline-parse guard correctly:
- ALLOWS writes when config is a valid empty {} (the Societus blind-spot fix)
- BLOCKS writes when config exists but is genuinely unparseable (garbage/truncated YAML)
- ALLOWS writes when config doesn't exist (new install, no file to parse)
- ALLOWS writes when config has valid non-empty content
"""

from __future__ import annotations

import logging
import os
from unittest.mock import patch

import pytest

from hermes_cli.config import save_config, set_config_value


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _config_path(tmp_path):
return tmp_path / "config.yaml"


def _write_config(tmp_path, content):
"""Write raw bytes to config.yaml (allows writing invalid YAML)."""
path = _config_path(tmp_path)
path.parent.mkdir(parents=True, exist_ok=True)
if isinstance(content, str):
content = content.encode("utf-8")
path.write_bytes(content)


def _read_config(tmp_path):
"""Read config.yaml content as bytes, or empty if not present."""
path = _config_path(tmp_path)
return path.read_bytes() if path.exists() else b""


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture(autouse=True)
def _isolated_hermes_home(tmp_path, monkeypatch):
"""Point HERMES_HOME at a temp dir so tests never touch real config."""
# Ensure the config path setup functions resolve to tmp_path
env_file = tmp_path / ".env"
env_file.touch()
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
yield tmp_path


# ---------------------------------------------------------------------------
# save_config parse guard tests
# ---------------------------------------------------------------------------

class TestSaveConfigParseGuard:
"""save_config must refuse to write when the existing file is unparseable,
but allow writes for valid empty {}, valid non-empty, and missing files."""

def test_valid_empty_dict_is_writable(self, caplog, _isolated_hermes_home):
"""A valid empty {} config must be writable (the Societus blind-spot fix)."""
_write_config(_isolated_hermes_home, b"{}\n")
caplog.set_level(logging.WARNING)

save_config({"model": {"provider": "test"}}, strip_defaults=False)
result = _read_config(_isolated_hermes_home)

assert "Refusing to save config" not in caplog.text
assert b"test" in result or b"provider" in result, (
"save_config should have written through a valid {} config"
)

def test_valid_non_empty_is_writable(self, caplog, _isolated_hermes_home):
"""A valid non-empty config must be writable."""
_write_config(_isolated_hermes_home, b"model:\n provider: existing\n")
caplog.set_level(logging.WARNING)

save_config({"model": {"provider": "updated"}}, strip_defaults=False)
result = _read_config(_isolated_hermes_home)

assert "Refusing to save config" not in caplog.text

def test_unparseable_garbage_is_blocked(self, caplog, _isolated_hermes_home):
"""An unparseable (garbage) config must be refused β€” guard fires."""
_write_config(_isolated_hermes_home, b"unclosed: [\n")
caplog.set_level(logging.WARNING)

# save_config will hit the guard and return early, so the file stays
save_config({"model": {"provider": "test"}}, strip_defaults=False)

assert "Refusing to save config" in caplog.text, (
"guard should log warning for unparseable config"
)
# File content must remain unchanged (the garbage)
assert _read_config(_isolated_hermes_home) == b"unclosed: [\n"

def test_missing_file_is_writable(self, caplog, _isolated_hermes_home):
"""A non-existent config file must be writable (new install)."""
assert not _config_path(_isolated_hermes_home).exists()
caplog.set_level(logging.WARNING)

save_config({"model": {"provider": "test"}}, strip_defaults=False)

assert "Refusing to save config" not in caplog.text
assert _config_path(_isolated_hermes_home).exists()


# ---------------------------------------------------------------------------
# set_config_value parse guard tests
# ---------------------------------------------------------------------------

class TestSetConfigValueParseGuard:
"""set_config_value must refuse when the existing file is unparseable,
but allow writes for valid and missing files."""

def test_valid_empty_dict_is_writable(self, caplog, _isolated_hermes_home):
"""set_config_value must write through a valid empty {} config."""
_write_config(_isolated_hermes_home, b"{}\n")
caplog.set_level(logging.WARNING)

set_config_value("model.provider", "test-provider")
result = _read_config(_isolated_hermes_home)

assert "Refusing to set config value" not in caplog.text
assert b"test-provider" in result

def test_valid_non_empty_is_writable(self, caplog, _isolated_hermes_home):
"""set_config_value must write through a valid non-empty config."""
_write_config(_isolated_hermes_home, b"model:\n provider: existing\n")
caplog.set_level(logging.WARNING)

set_config_value("model.provider", "updated")
result = _read_config(_isolated_hermes_home)

assert "Refusing to set config value" not in caplog.text
assert b"updated" in result

def test_unparseable_garbage_is_blocked(self, caplog, _isolated_hermes_home):
"""set_config_value must refuse when existing config is unparseable."""
_write_config(_isolated_hermes_home, b"unclosed: [\n")
caplog.set_level(logging.WARNING)

set_config_value("model.provider", "test")

assert "Refusing to set config value" in caplog.text, (
"guard should log warning for unparseable config"
)
assert _read_config(_isolated_hermes_home) == b"unclosed: [\n"

def test_missing_file_is_writable(self, caplog, _isolated_hermes_home):
"""set_config_value must write when no config file exists."""
assert not _config_path(_isolated_hermes_home).exists()
caplog.set_level(logging.WARNING)

set_config_value("model.provider", "test")
assert "Refusing to set config value" not in caplog.text


# ---------------------------------------------------------------------------
# Non-mapping root tests (Teknium review: list / scalar root must also refuse)
# ---------------------------------------------------------------------------

class TestNonMappingRootGuard:
"""A top-level list (``[]``) or scalar parses WITHOUT raising, so an
exception-only guard would let save_config / set_config_value wipe the
file. read_raw_config() coerces both to {} at config.py:6931, so the
guard must type-check the parsed result, not just catch parse errors.
"""

def test_list_root_save_config_blocked(self, caplog, _isolated_hermes_home):
"""A list-root config must be refused on save_config (not coerced to {})."""
_write_config(_isolated_hermes_home, b"- a\n- b\n")
caplog.set_level(logging.WARNING)

save_config({"model": {"provider": "test"}}, strip_defaults=False)

assert "Refusing to save config" in caplog.text, (
"list-root config should be refused"
)
assert _read_config(_isolated_hermes_home) == b"- a\n- b\n"

def test_list_root_set_config_value_blocked(self, caplog, _isolated_hermes_home):
"""A list-root config must be refused on set_config_value."""
_write_config(_isolated_hermes_home, b"- a\n- b\n")
caplog.set_level(logging.WARNING)

set_config_value("model.provider", "test")

assert "Refusing to set config value" in caplog.text, (
"list-root config should be refused"
)
assert _read_config(_isolated_hermes_home) == b"- a\n- b\n"

def test_scalar_root_save_config_blocked(self, caplog, _isolated_hermes_home):
"""A scalar-root config must be refused on save_config."""
_write_config(_isolated_hermes_home, b"just a string\n")
caplog.set_level(logging.WARNING)

save_config({"model": {"provider": "test"}}, strip_defaults=False)

assert "Refusing to save config" in caplog.text, (
"scalar-root config should be refused"
)
assert _read_config(_isolated_hermes_home) == b"just a string\n"

def test_scalar_root_set_config_value_blocked(self, caplog, _isolated_hermes_home):
"""A scalar-root config must be refused on set_config_value."""
_write_config(_isolated_hermes_home, b"42\n")
caplog.set_level(logging.WARNING)

set_config_value("model.provider", "test")

assert "Refusing to set config value" in caplog.text, (
"scalar-root config should be refused"
)
assert _read_config(_isolated_hermes_home) == b"42\n"


# ---------------------------------------------------------------------------
# Auth provider writer guard (Teknium review: cover auth.py write path too)
# ---------------------------------------------------------------------------

class TestAuthProviderWriterGuard:
"""_update_config_for_provider must route through the same shared guard,
so a malformed / non-mapping existing config is not clobbered by the
provider write either.
"""

def test_unparseable_blocked(self, caplog, _isolated_hermes_home):
_write_config(_isolated_hermes_home, b"unclosed: [\n")
caplog.set_level(logging.WARNING)

from hermes_cli.auth import _update_config_for_provider

_update_config_for_provider("zai", "", default_model="glm-5.2")

assert "Refusing to set provider config" in caplog.text, (
"auth provider writer should refuse unparseable config"
)
assert _read_config(_isolated_hermes_home) == b"unclosed: [\n"

def test_list_root_blocked(self, caplog, _isolated_hermes_home):
_write_config(_isolated_hermes_home, b"- a\n- b\n")
caplog.set_level(logging.WARNING)

from hermes_cli.auth import _update_config_for_provider

_update_config_for_provider("zai", "", default_model="glm-5.2")

assert "Refusing to set provider config" in caplog.text, (
"auth provider writer should refuse list-root config"
)
assert _read_config(_isolated_hermes_home) == b"- a\n- b\n"

def test_valid_empty_writes(self, caplog, _isolated_hermes_home):
"""A valid empty {} config must be writable through the auth writer."""
_write_config(_isolated_hermes_home, b"{}\n")
caplog.set_level(logging.WARNING)

from hermes_cli.auth import _update_config_for_provider

_update_config_for_provider("zai", "", default_model="glm-5.2")

assert "Refusing to set provider config" not in caplog.text
# The written model section should carry the provider + default.
assert b"glm-5.2" in _read_config(_isolated_hermes_home)
Loading