diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 91523d76f8ee..da49a695180a 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -6253,6 +6253,13 @@ def _resolve_task_provider_model( cfg_model = str(task_config.get("model", "")).strip() or None cfg_base_url = str(task_config.get("base_url", "")).strip() or None cfg_api_key = str(task_config.get("api_key", "")).strip() or None + # Resolve key_env → env var when api_key is not set directly + if not cfg_api_key: + cfg_key_env = str( + task_config.get("key_env") or task_config.get("api_key_env") or "" + ).strip() + if cfg_key_env: + cfg_api_key = os.getenv(cfg_key_env, "").strip() or None cfg_api_mode = str(task_config.get("api_mode", "")).strip() or None # 'auto' is a sentinel meaning "inherit from main runtime / auto-detect", not diff --git a/agent/system_prompt.py b/agent/system_prompt.py index 41337e4e9fbf..aab34cb799e1 100644 --- a/agent/system_prompt.py +++ b/agent/system_prompt.py @@ -46,6 +46,7 @@ from agent.prompt_builder import ( drain_truncation_warnings, ) from agent.runtime_cwd import resolve_context_cwd +from hermes_constants import get_hermes_home from utils import is_truthy_value @@ -395,7 +396,7 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None) if active_profile == "default": stable_parts.append( "Active Hermes profile: default. Other profiles (if any) live " - "under ~/.hermes/profiles//. Each profile has its own " + "under " + str(get_hermes_home()) + "/profiles//. Each profile has its own " "skills/, plugins/, cron/, and memories/ that affect a different " "session than this one. Do not modify another profile's " "skills/plugins/cron/memories unless the user explicitly directs " @@ -404,9 +405,9 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None) else: stable_parts.append( f"Active Hermes profile: {active_profile}. This session reads " - f"and writes ~/.hermes/profiles/{active_profile}/. The default " - f"profile's data lives at ~/.hermes/skills/, ~/.hermes/plugins/, " - f"~/.hermes/cron/, ~/.hermes/memories/ — those belong to a " + f"and writes {get_hermes_home()}/profiles/{active_profile}/. The default " + f"profile's data lives at {get_hermes_home()}/skills/, {get_hermes_home()}/plugins/, " + f"{get_hermes_home()}/cron/, {get_hermes_home()}/memories/ — those belong to a " f"different session run from a different shell. Do NOT modify " f"another profile's skills/plugins/cron/memories unless the user " f"explicitly directs you to. The cross-profile write guard will " diff --git a/hermes_cli/config.py b/hermes_cli/config.py index fc9d75470aeb..f29ce777427b 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -7829,11 +7829,16 @@ def _quote_env_value(value: str) -> str: """Quote .env values containing characters with special dotenv meaning.""" if value == "": return value + # Internal whitespace (space/tab/etc.) must be quoted so shell `set -a; . file` + # word-splits don't break paths like macOS "Application Support". Leading/ + # trailing whitespace is already covered by the strip check; any() covers + # internal runs that strip() would leave alone. needs_quoting = ( "#" in value or '"' in value or "'" in value or value != value.strip() + or any(c.isspace() for c in value) ) if not needs_quoting: return value diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index 3ea53f7aaa7c..9dea6e5fe924 100644 --- a/hermes_cli/env_loader.py +++ b/hermes_cli/env_loader.py @@ -2,6 +2,8 @@ from __future__ import annotations +import codecs +import io import os import sys from pathlib import Path @@ -21,6 +23,12 @@ _CREDENTIAL_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_KEY") # tests) don't spam the same warning multiple times. _WARNED_KEYS: set[str] = set() +# Paths we've already emitted a UTF-32 refuse-to-mangle warning for. +# load_hermes_dotenv can call _sanitize_env_file_if_needed multiple times +# for the same file (user env + project env + hot-reload); once per path +# is enough. +_WARNED_UTF32_PATHS: set[str] = set() + # Map of env-var name → source label ("bitwarden", etc.) for credentials # that were injected by an external secret source during load_hermes_dotenv(). # Used by setup / `hermes model` flows to label detected credentials so @@ -176,6 +184,12 @@ def _sanitize_env_file_if_needed(path: Path) -> None: with ``ValueError: embedded null byte`` — typically introduced by copy-pasting API keys from terminals or rich-text editors. + Encoding: sniffs a leading BOM *before* any text decode. UTF-16 + (Notepad "Unicode") is decoded correctly and rewritten as clean + UTF-8. UTF-32 is refused (left untouched) so we never fall through + to the errors=replace corruption path. Order of BOM checks matters: + UTF-32-LE's BOM starts with UTF-16-LE's FF FE. + We delegate to ``hermes_cli.config._sanitize_env_lines`` which already knows all valid Hermes env-var names and can split concatenated lines correctly. @@ -187,16 +201,70 @@ def _sanitize_env_file_if_needed(path: Path) -> None: except ImportError: return # early bootstrap — config module not available yet - read_kw = {"encoding": "utf-8-sig", "errors": "replace"} try: - with open(path, **read_kw) as f: - original = f.readlines() + raw = path.read_bytes() + except Exception: + return + + # Sniff leading BOM bytes BEFORE decoding. ORDER MATTERS: + # codecs.BOM_UTF32_LE is FF FE 00 00, which startswith + # codecs.BOM_UTF16_LE (FF FE). Checking UTF-16 first would + # misdetect UTF-32-LE as UTF-16-LE and mangle the file. + force_utf8_rewrite = False + if raw.startswith(codecs.BOM_UTF32_LE) or raw.startswith(codecs.BOM_UTF32_BE): + # Lazy import keeps the module import block identical to #65124's + # codecs/io additions so the two PRs auto-merge either order. + path_key = str(path.resolve()) + if path_key not in _WARNED_UTF32_PATHS: + _WARNED_UTF32_PATHS.add(path_key) + import logging + + logging.getLogger(__name__).warning( + "Skipping .env sanitize for %s: UTF-32 BOM detected; " + "leaving file untouched to avoid corruption", + path, + ) + return + if raw.startswith(codecs.BOM_UTF16_LE) or raw.startswith(codecs.BOM_UTF16_BE): + # "utf-16" uses the BOM to select endianness and strips it. + # TextIOWrapper + newline=None matches open()'s universal-newlines + # line splitting (\\n/\\r\\n/\\r only — not splitlines()'s extra + # Unicode boundaries like U+2028), so sanitize sees the same lines + # as the UTF-8 path. + try: + with io.TextIOWrapper( + io.BytesIO(raw), encoding="utf-16", newline=None + ) as f: + original = f.readlines() + except UnicodeDecodeError: + return + # Source is UTF-16 on disk; always rewrite as clean UTF-8 so + # the subsequent utf-8 dotenv load sees a canonical file. + force_utf8_rewrite = True + else: + # Default path: utf-8-sig (strips UTF-8 BOM if present) with + # errors=replace so embedded NULs can be stripped below. + try: + with open(path, encoding="utf-8-sig", errors="replace") as f: + original = f.readlines() + except Exception: + return + # Defense-in-depth: errors=replace turns undecodable leading + # bytes into U+FFFD. Persisting that glues replacement chars + # onto the first key name and rewrites the file permanently + # (the UTF-16-with-BOM corruption path before BOM sniffing). + # Leave the file untouched rather than write the mangling. + if original and original[0].startswith("\ufffd"): + return + + try: # Strip null bytes before _sanitize_env_lines so they never # reach python-dotenv (which passes them to os.environ and - # crashes with ValueError). + # crashes with ValueError). Also intentionally repairs + # BOM-less UTF-16 (NUL-padded ASCII) into clean UTF-8. stripped = [line.replace("\x00", "") for line in original] sanitized = _sanitize_env_lines(stripped) - if sanitized != original: + if sanitized != original or force_utf8_rewrite: import tempfile fd, tmp = tempfile.mkstemp( dir=str(path.parent), suffix=".tmp", prefix=".env_" diff --git a/tests/hermes_cli/test_config.py b/tests/hermes_cli/test_config.py index cc1d0d361617..1d9c396fda60 100644 --- a/tests/hermes_cli/test_config.py +++ b/tests/hermes_cli/test_config.py @@ -607,6 +607,210 @@ class TestSaveEnvValueSecure: assert parsed["ANTHROPIC_TOKEN"] == token assert load_env()["ANTHROPIC_TOKEN"] == token + def test_save_env_value_quotes_values_with_internal_spaces(self, tmp_path): + """Internal spaces must be quoted so shell-sourcing does not word-split. + + Sibling of installer #57247: core writer left + TERMINAL_SSH_KEY=/Users/.../Application Support/... unquoted. + python-dotenv still parsed it; ``set -a; . file`` failed. + """ + import subprocess + from dotenv import dotenv_values + + path = "/Users/paulo/Library/Application Support/hermes/keys/id_ed25519" + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}, clear=False): + os.environ.pop("TERMINAL_SSH_KEY", None) + save_env_value("TERMINAL_SSH_KEY", path) + + env_path = tmp_path / ".env" + content = env_path.read_text(encoding="utf-8") + assert f'TERMINAL_SSH_KEY="{path}"' in content + + parsed = dotenv_values(str(env_path)) + assert parsed["TERMINAL_SSH_KEY"] == path + assert load_env()["TERMINAL_SSH_KEY"] == path + + # Shell source must round-trip (this is what the bug broke). + r = subprocess.run( + [ + "env", + "-i", + "sh", + "-c", + f"set -a; . '{env_path}'; set +a; " + f'printf "%s" "$TERMINAL_SSH_KEY"', + ], + capture_output=True, + text=True, + ) + assert r.returncode == 0, r.stderr + assert r.stderr == "" + assert r.stdout == path + + def test_save_env_value_quotes_values_with_tabs(self, tmp_path): + """Tabs trigger quoting; round-trip via dotenv and shell source.""" + import subprocess + from dotenv import dotenv_values + + value = "left\tright" + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}, clear=False): + os.environ.pop("TABBY_KEY", None) + save_env_value("TABBY_KEY", value) + + env_path = tmp_path / ".env" + content = env_path.read_text(encoding="utf-8") + assert f'TABBY_KEY="{value}"' in content + + parsed = dotenv_values(str(env_path)) + assert parsed["TABBY_KEY"] == value + assert load_env()["TABBY_KEY"] == value + + r = subprocess.run( + [ + "env", + "-i", + "sh", + "-c", + f"set -a; . '{env_path}'; set +a; " + f'printf "%s" "$TABBY_KEY"', + ], + capture_output=True, + text=True, + ) + assert r.returncode == 0, r.stderr + assert r.stderr == "" + assert r.stdout == value + + def test_save_env_value_spaced_path_is_idempotent(self, tmp_path): + """Saving the same spaced value twice must not grow quotes.""" + path = "/Users/me/Application Support/key" + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}, clear=False): + os.environ.pop("TERMINAL_SSH_KEY", None) + save_env_value("TERMINAL_SSH_KEY", path) + first = (tmp_path / ".env").read_text(encoding="utf-8") + save_env_value("TERMINAL_SSH_KEY", path) + second = (tmp_path / ".env").read_text(encoding="utf-8") + + assert first == second + assert first.count('TERMINAL_SSH_KEY="') == 1 + assert '""' not in first + assert f'TERMINAL_SSH_KEY="{path}"' in first + + def test_save_env_value_readback_resave_is_idempotent(self, tmp_path): + """hermes setup path: dotenv unquotes, then re-save must not grow quotes.""" + from dotenv import dotenv_values + + path = "/Users/me/Application Support/key" + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}, clear=False): + os.environ.pop("TERMINAL_SSH_KEY", None) + save_env_value("TERMINAL_SSH_KEY", path) + first = (tmp_path / ".env").read_text(encoding="utf-8") + + # Real read-back boundary (what setup uses via get_env_value/dotenv). + read_back = dotenv_values(str(tmp_path / ".env"))["TERMINAL_SSH_KEY"] + assert read_back == path + save_env_value("TERMINAL_SSH_KEY", read_back) + second = (tmp_path / ".env").read_text(encoding="utf-8") + + assert first == second + assert f'TERMINAL_SSH_KEY="{path}"' in second + + def test_save_env_value_strips_newlines_before_quoting(self, tmp_path): + """save_env_value strips \\n/\\r before _quote_env_value; result is one line. + + Pins the boundary so any(c.isspace()) never quotes multi-line dotenv + values through this writer (newlines never reach the quoter). + """ + from dotenv import dotenv_values + + raw = "line1\nline2\rline3" + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}, clear=False): + os.environ.pop("MULTI_KEY", None) + save_env_value("MULTI_KEY", raw) + + content = (tmp_path / ".env").read_text(encoding="utf-8") + # Single KEY= line, no embedded raw newlines in the value payload. + lines = [ln for ln in content.splitlines() if ln.startswith("MULTI_KEY=")] + assert len(lines) == 1 + assert "\n" not in lines[0] + assert "\r" not in lines[0] + # Newlines stripped -> "line1line2line3" has no whitespace -> unquoted. + assert lines[0] == "MULTI_KEY=line1line2line3" + parsed = dotenv_values(str(tmp_path / ".env")) + assert parsed["MULTI_KEY"] == "line1line2line3" + + def test_save_env_value_simple_values_stay_unquoted(self, tmp_path): + """No quoting churn: plain values remain bare; untouched lines unchanged.""" + env_path = tmp_path / ".env" + # Pre-existing lines: one simple, one already correctly bare. + env_path.write_text( + "KEEP_SIMPLE=plainvalue\n" + "OTHER_KEY=foo123\n", + encoding="utf-8", + ) + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}, clear=False): + os.environ.pop("NEW_KEY", None) + os.environ.pop("KEEP_SIMPLE", None) + save_env_value("NEW_KEY", "bar-simple") + + content = env_path.read_text(encoding="utf-8") + # Newly written simple value is unquoted. + assert "NEW_KEY=bar-simple\n" in content + assert 'NEW_KEY="' not in content + # Untouched pre-existing simple lines are not re-quoted. + assert "KEEP_SIMPLE=plainvalue\n" in content + assert "OTHER_KEY=foo123\n" in content + assert 'KEEP_SIMPLE="' not in content + assert 'OTHER_KEY="' not in content + + def test_save_env_value_does_not_requote_untouched_spaced_lines(self, tmp_path): + """Mass-requote guard: rewriting another key leaves legacy spaced + lines as-is (fix only applies when that key is saved again). + """ + env_path = tmp_path / ".env" + legacy = ( + "TERMINAL_SSH_KEY=/Users/me/Application Support/key\n" + "PLAIN=ok\n" + ) + env_path.write_text(legacy, encoding="utf-8") + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}, clear=False): + os.environ.pop("PLAIN", None) + save_env_value("PLAIN", "ok2") + + content = env_path.read_text(encoding="utf-8") + # Legacy spaced line not re-serialized by this write. + assert ( + "TERMINAL_SSH_KEY=/Users/me/Application Support/key\n" in content + ) + assert 'TERMINAL_SSH_KEY="' not in content + assert "PLAIN=ok2\n" in content + + def test_save_env_value_already_quoted_input_is_not_double_wrapped_idempotently( + self, tmp_path + ): + """Callers pass raw values; if a value literally contains quote + characters, escaping+wrap is the dialect (#57249). Re-saving the + same raw value is stable (no quote growth). load_env round-trips. + """ + # User-typed value that already includes surrounding quotes as data. + raw = '"/Users/me/Application Support/key"' + with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}, clear=False): + os.environ.pop("TERMINAL_SSH_KEY", None) + save_env_value("TERMINAL_SSH_KEY", raw) + first = (tmp_path / ".env").read_text(encoding="utf-8") + save_env_value("TERMINAL_SSH_KEY", raw) + second = (tmp_path / ".env").read_text(encoding="utf-8") + assert first == second + # One outer wrap layer only (escaped inner quotes, not nested wraps). + line = [ + ln for ln in first.splitlines() if ln.startswith("TERMINAL_SSH_KEY=") + ][0] + assert line.startswith('TERMINAL_SSH_KEY="') + assert line.endswith('"') + assert line.count('TERMINAL_SSH_KEY="') == 1 + # Escaping dialect end-to-end: load sees the raw input, not stripped quotes. + assert load_env()["TERMINAL_SSH_KEY"] == raw + class TestRemoveEnvValue: def test_removes_key_from_env_file(self, tmp_path): @@ -726,7 +930,7 @@ class TestSaveConfigAtomicity: # Read raw YAML to verify it's valid and correct config_path = tmp_path / "config.yaml" - with open(config_path) as f: + with open(config_path, encoding="utf-8") as f: raw = yaml.safe_load(f) assert raw["model"] == "test/atomic-model" assert raw["agent"]["max_turns"] == 77 diff --git a/tests/hermes_cli/test_env_loader.py b/tests/hermes_cli/test_env_loader.py index f719fae01ce7..a5a2ee752726 100644 --- a/tests/hermes_cli/test_env_loader.py +++ b/tests/hermes_cli/test_env_loader.py @@ -1,3 +1,4 @@ +import codecs import importlib import os import sys @@ -103,3 +104,268 @@ def test_main_import_applies_user_env_over_shell_values(tmp_path, monkeypatch): assert os.getenv("OPENAI_BASE_URL") == "https://new.example/v1" assert os.getenv("HERMES_INFERENCE_PROVIDER") == "custom" + + +# --------------------------------------------------------------------------- +# UTF-16 / UTF-32 .env sanitizer coverage +# +# Scope note: intentionally NO UTF-8-BOM assertions here. UTF-8 BOM handling +# for _load_dotenv_with_fallback is #65124's un-merged fix; a test here would +# couple the PRs. This suite covers only the sanitizer rewrite path for +# UTF-16/32 (and UTF-8 / cp1252 regression guards for that path). +# --------------------------------------------------------------------------- + + +def _assert_clean_utf8_env_on_disk(env_file, *, first_key: str) -> None: + """On-disk file must be clean UTF-8: no BOM, no U+FFFD, canonical key.""" + after = env_file.read_bytes() + assert not after.startswith(codecs.BOM_UTF8) + assert not after.startswith(codecs.BOM_UTF16_LE) + assert not after.startswith(codecs.BOM_UTF16_BE) + text = after.decode("utf-8") # strict — raises if not clean UTF-8 + assert "\ufffd" not in text + assert text.startswith(f"{first_key}=") or f"\n{first_key}=" in text + assert first_key.encode("ascii") in after + + +def test_utf16_le_bom_env_loads_and_rewrites_clean_utf8(tmp_path, monkeypatch): + """Notepad 'Unicode' (UTF-16-LE + BOM): first key loads; file rewritten UTF-8.""" + home = tmp_path / "hermes" + home.mkdir() + env_file = home / ".env" + content = "HERMES_TEST_KEY=hello_utf16\nSECOND_KEY=world\n" + env_file.write_bytes(codecs.BOM_UTF16_LE + content.encode("utf-16-le")) + + monkeypatch.delenv("HERMES_TEST_KEY", raising=False) + monkeypatch.delenv("SECOND_KEY", raising=False) + monkeypatch.delenv("\ufffd\ufffdHERMES_TEST_KEY", raising=False) + + loaded = load_hermes_dotenv(hermes_home=home) + + assert loaded == [env_file] + assert os.getenv("HERMES_TEST_KEY") == "hello_utf16" + assert os.getenv("SECOND_KEY") == "world" + assert os.environ.get("\ufffd\ufffdHERMES_TEST_KEY") is None + _assert_clean_utf8_env_on_disk(env_file, first_key="HERMES_TEST_KEY") + + +def test_utf16_be_bom_env_loads_and_rewrites_clean_utf8(tmp_path, monkeypatch): + """UTF-16-BE + BOM: first key loads; file rewritten as clean UTF-8.""" + home = tmp_path / "hermes" + home.mkdir() + env_file = home / ".env" + content = "HERMES_TEST_KEY=hello_utf16\nSECOND_KEY=world\n" + env_file.write_bytes(codecs.BOM_UTF16_BE + content.encode("utf-16-be")) + + monkeypatch.delenv("HERMES_TEST_KEY", raising=False) + monkeypatch.delenv("SECOND_KEY", raising=False) + + loaded = load_hermes_dotenv(hermes_home=home) + + assert loaded == [env_file] + assert os.getenv("HERMES_TEST_KEY") == "hello_utf16" + assert os.getenv("SECOND_KEY") == "world" + _assert_clean_utf8_env_on_disk(env_file, first_key="HERMES_TEST_KEY") + + +def test_utf16_le_no_bom_still_repairs_to_utf8(tmp_path, monkeypatch): + """BOM-less UTF-16-LE: NUL-strip repair is now intentional; rewrites UTF-8.""" + home = tmp_path / "hermes" + home.mkdir() + env_file = home / ".env" + content = "HERMES_TEST_KEY=hello_utf16\nSECOND_KEY=world\n" + env_file.write_bytes(content.encode("utf-16-le")) # no BOM + + monkeypatch.delenv("HERMES_TEST_KEY", raising=False) + monkeypatch.delenv("SECOND_KEY", raising=False) + + loaded = load_hermes_dotenv(hermes_home=home) + + assert loaded == [env_file] + assert os.getenv("HERMES_TEST_KEY") == "hello_utf16" + assert os.getenv("SECOND_KEY") == "world" + _assert_clean_utf8_env_on_disk(env_file, first_key="HERMES_TEST_KEY") + + +def test_utf16_be_no_bom_still_repairs_to_utf8(tmp_path, monkeypatch): + """BOM-less UTF-16-BE: NULs are on the opposite side; still repairs.""" + home = tmp_path / "hermes" + home.mkdir() + env_file = home / ".env" + content = "HERMES_TEST_KEY=hello_utf16\nSECOND_KEY=world\n" + env_file.write_bytes(content.encode("utf-16-be")) # no BOM + + monkeypatch.delenv("HERMES_TEST_KEY", raising=False) + monkeypatch.delenv("SECOND_KEY", raising=False) + + loaded = load_hermes_dotenv(hermes_home=home) + + assert loaded == [env_file] + assert os.getenv("HERMES_TEST_KEY") == "hello_utf16" + assert os.getenv("SECOND_KEY") == "world" + _assert_clean_utf8_env_on_disk(env_file, first_key="HERMES_TEST_KEY") + + +def test_utf16_le_bom_preserves_non_ascii_values(tmp_path, monkeypatch): + """UTF-16-LE+BOM rewrite must preserve non-ASCII values (not just ASCII keys). + + Uses non-credential var names so _sanitize_loaded_credentials does not + strip non-ASCII from values (that path only targets *_KEY/*_TOKEN/etc.). + """ + home = tmp_path / "hermes" + home.mkdir() + env_file = home / ".env" + content = "GREETING=café\nCJK_LABEL=日本語\n" + env_file.write_bytes(codecs.BOM_UTF16_LE + content.encode("utf-16-le")) + + monkeypatch.delenv("GREETING", raising=False) + monkeypatch.delenv("CJK_LABEL", raising=False) + + loaded = load_hermes_dotenv(hermes_home=home) + + assert loaded == [env_file] + assert os.getenv("GREETING") == "café" + assert os.getenv("CJK_LABEL") == "日本語" + after = env_file.read_bytes() + assert after.decode("utf-8") # strict + assert "café".encode("utf-8") in after + assert "日本語".encode("utf-8") in after + assert b"\xef\xbf\xbd" not in after + + +def test_utf32_le_bom_leaves_file_untouched(tmp_path, caplog): + """UTF-32-LE BOM: refuse-to-mangle (leave bytes untouched + warning). + + UTF-32-LE's BOM starts with UTF-16-LE's FF FE; sniff order must check + UTF-32 first so we never misdetect and corrupt. + + Exercises ``_sanitize_env_file_if_needed`` only: the dotenv load path + is out of scope here (#65124's surface) and still cannot ingest UTF-32. + """ + import logging + + from hermes_cli.env_loader import _sanitize_env_file_if_needed + + env_file = tmp_path / ".env" + content = "HERMES_TEST_KEY=hello_utf32\nSECOND_KEY=world\n" + raw = codecs.BOM_UTF32_LE + content.encode("utf-32-le") + env_file.write_bytes(raw) + + with caplog.at_level(logging.WARNING, logger="hermes_cli.env_loader"): + _sanitize_env_file_if_needed(env_file) + + assert env_file.read_bytes() == raw # untouched + assert any("UTF-32" in r.message for r in caplog.records) + + +def test_utf32_be_bom_leaves_file_untouched(tmp_path, caplog): + """UTF-32-BE BOM: same refuse-to-mangle path as LE (ordering independence).""" + import logging + + from hermes_cli.env_loader import _sanitize_env_file_if_needed + + env_file = tmp_path / ".env" + content = "HERMES_TEST_KEY=hello_utf32\nSECOND_KEY=world\n" + raw = codecs.BOM_UTF32_BE + content.encode("utf-32-be") + env_file.write_bytes(raw) + + with caplog.at_level(logging.WARNING, logger="hermes_cli.env_loader"): + _sanitize_env_file_if_needed(env_file) + + assert env_file.read_bytes() == raw + assert any("UTF-32" in r.message for r in caplog.records) + + +def test_utf32_warning_fires_once_per_path(tmp_path, caplog, monkeypatch): + """Three sanitize calls on the same UTF-32 file → exactly one warning. + + Matches house style for warn-once (module-level seen-set, same class as + ``_WARNED_KEYS``): hot-reload / multi-entry load must not spam logs. + """ + import logging + + import hermes_cli.env_loader as env_loader + from hermes_cli.env_loader import _sanitize_env_file_if_needed + + # Isolate process-level seen-set so other tests' paths don't leak in. + monkeypatch.setattr(env_loader, "_WARNED_UTF32_PATHS", set()) + + env_file = tmp_path / ".env" + content = "HERMES_TEST_KEY=hello_utf32\nSECOND_KEY=world\n" + raw = codecs.BOM_UTF32_LE + content.encode("utf-32-le") + env_file.write_bytes(raw) + + with caplog.at_level(logging.WARNING, logger="hermes_cli.env_loader"): + _sanitize_env_file_if_needed(env_file) + _sanitize_env_file_if_needed(env_file) + _sanitize_env_file_if_needed(env_file) + + utf32_warnings = [r for r in caplog.records if "UTF-32" in r.message] + assert len(utf32_warnings) == 1 + assert env_file.read_bytes() == raw + + +def test_leading_replacement_char_does_not_rewrite(tmp_path): + """errors=replace FFFD-on-first-line guard: do not persist mangling. + + Leading 0xFF is not a UTF-16/32 BOM (those need the second BOM byte) but + is undecodable as UTF-8, so the replace path would glue U+FFFD onto the + key. The guard must leave the on-disk bytes untouched. + """ + from hermes_cli.env_loader import _sanitize_env_file_if_needed + + env_file = tmp_path / ".env" + raw = b"\xffHERMES_TEST_KEY=should-not-rewrite\nSECOND_KEY=ok\n" + env_file.write_bytes(raw) + + _sanitize_env_file_if_needed(env_file) + + assert env_file.read_bytes() == raw + + +def test_plain_utf8_env_regression(tmp_path, monkeypatch): + """Plain UTF-8 .env must keep loading after the UTF-16 sanitize changes.""" + home = tmp_path / "hermes" + home.mkdir() + env_file = home / ".env" + before = b"OPENAI_API_KEY=sk-plain\nSECOND_KEY=ok\n" + env_file.write_bytes(before) + + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + monkeypatch.delenv("SECOND_KEY", raising=False) + + loaded = load_hermes_dotenv(hermes_home=home) + + assert loaded == [env_file] + assert os.getenv("OPENAI_API_KEY") == "sk-plain" + assert os.getenv("SECOND_KEY") == "ok" + # No spurious rewrite of an already-clean file. + assert env_file.read_bytes() == before + + +def test_cp1252_env_regression_does_not_crash(tmp_path, monkeypatch): + """cp1252/latin-1 body must not crash sanitize; ASCII keys still usable. + + 0xE9 is 'é' in cp1252 and incomplete as UTF-8. First line does not begin + with U+FFFD, so the FFFD guard must not refuse the whole file. + + Sanitize leaves the file bytes alone when the only "change" is + errors=replace on values (original already replace-decoded equals + sanitized), so _load_dotenv_with_fallback's latin-1 path recovers café. + """ + home = tmp_path / "hermes" + home.mkdir() + env_file = home / ".env" + before = b"ASCII_KEY=ok\nLATIN1_VALUE=caf\xe9\n" + env_file.write_bytes(before) + + monkeypatch.delenv("ASCII_KEY", raising=False) + monkeypatch.delenv("LATIN1_VALUE", raising=False) + + loaded = load_hermes_dotenv(hermes_home=home) + + assert loaded == [env_file] + assert os.getenv("ASCII_KEY") == "ok" + assert os.getenv("LATIN1_VALUE") == "café" + # Sanitize must not have rewritten (would have persisted U+FFFD). + assert env_file.read_bytes() == before