mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-21 16:18:55 +00:00
Merge pull request #67192 from NousResearch/bb/p2-config-salvage
fix(config): P2 batch — .env quoting/UTF-16, aux key_env, profile-aware system prompt
This commit is contained in:
brooklyn!
GitHub
commit
34e66a0d52
6 changed files with 561 additions and 10 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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/<name>/. Each profile has its own "
|
||||
"under " + str(get_hermes_home()) + "/profiles/<name>/. 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 "
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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_"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue