diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index 3ea53f7aaa7c..8018280d5b98 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 @@ -176,6 +178,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 +195,67 @@ 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. + 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_env_loader.py b/tests/hermes_cli/test_env_loader.py index f719fae01ce7..719b272a5e4c 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,239 @@ 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_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