mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 16:18:55 +00:00
fix(cli): sanitize UTF-16 .env without corrupting the first key
Notepad "Unicode" saves write UTF-16 with a BOM. The sanitizer decoded those bytes as utf-8-sig with errors=replace, glued U+FFFD onto the first key name, stripped NULs, and rewrote the mangled content permanently. Sniff leading BOMs before any text decode (UTF-32 before UTF-16, because UTF-32-LE's BOM starts with UTF-16-LE's FF FE). Decode UTF-16 correctly and rewrite as clean UTF-8. Refuse UTF-32 (leave untouched + warning). After errors=replace, do not persist a first line that starts with U+FFFD. Does not touch _load_dotenv_with_fallback (#65124's surface).
This commit is contained in:
Paulo Nascimento
Brooklyn Nicholson
parent
4441e11c77
commit
7d597cc5d4
2 changed files with 301 additions and 5 deletions
|
|
@ -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_"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue