refactor(env): agent.secret_scope.load_env_file is the only .env tokenizer; six hand parsers collapse onto it
Six independent line-parsers with three different quoting/comment semantics
read the same .env files: tools/skills_tool.load_env (strip("\"'"), no inline
comments), hermes_cli/managed_scope._parse_env (same, no export, no BOM),
web_server_cron._profile_env_value (plain utf-8, no BOM), profile_cmd
._env_file_has_key, env_loader._env_keys_defined_in_dotenv (utf-8, so a BOM'd
first key stayed "\ufeffKEY" and the dashboard profile scrub missed line 1),
mem0/_setup._prompt_api_key (startswith scan, no quote strip). The boundary
parsers (scrub key set, skill secret capture) therefore disagreed with the
parser that installs the profile scope.
Now every one is a 1-3 line forwarder onto load_env_file, and
hermes_cli.config.load_env is memo over it (public signature unchanged).
_parse_env_value moves next to its only caller in secret_scope.
load_env_file gains the same latin-1 fallback env_loader uses to install
into os.environ, so a mis-encoded file yields the same key set on both sides.
Managed .env keeps its fail-LOUD contract (decode error logs and ignores the
file) instead of load_env_file's fail-soft {}.
Behavior change: managed .env, skills_tool and mem0 setup now honour
`export`, quoted-value escapes and inline comments the way the profile scope
does; web_server_cron and the dashboard scrub tolerate a BOM.
Invariant test: a BOM'd/export/quoted/commented .env yields the same key set
via load_hermes_dotenv (installer), load_env_file (scope) and
_env_keys_defined_in_dotenv (scrub); fails with the old scrub parser.
This commit is contained in:
+40
-7
@@ -11,6 +11,7 @@ falling back to ``os.environ``. Design: ``docs/design/multiplexing-gateway.md``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import codecs
|
||||
import os
|
||||
import re
|
||||
from contextvars import ContextVar, Token
|
||||
@@ -137,6 +138,13 @@ def get_secret(name: str, default: Optional[str] = None) -> Optional[str]:
|
||||
return _environ_or(name, default)
|
||||
|
||||
|
||||
def get_secret_str(name: str, default: str = "") -> str:
|
||||
"""``get_secret`` for callers that want a ``str``: ``default`` only when the secret is genuinely
|
||||
unset. Still raises ``UnscopedSecretError`` — swallowing it hides a spawn-site bug."""
|
||||
val = get_secret(name, default)
|
||||
return default if val is None else val
|
||||
|
||||
|
||||
def _strip_inline_comment(value: str) -> str:
|
||||
"""Strip a dotenv-style inline comment (python-dotenv semantics): quoted values
|
||||
scan to the matching close quote (backslash-aware for double quotes) and drop a
|
||||
@@ -160,17 +168,42 @@ def _strip_inline_comment(value: str) -> str:
|
||||
return re.split(r"\s+#", value, maxsplit=1)[0].strip()
|
||||
|
||||
|
||||
def _parse_env_value(raw_value: str) -> str:
|
||||
"""Parse the small .env value subset Hermes writes itself (bare, 'single', or "double" with
|
||||
``\\"`` / ``\\\\`` escapes)."""
|
||||
value = raw_value.strip()
|
||||
if len(value) >= 2 and value[0] == value[-1] == '"':
|
||||
quoted = value[1:-1]
|
||||
parsed: list[str] = []
|
||||
i = 0
|
||||
while i < len(quoted):
|
||||
escaped = quoted[i] == "\\" and quoted[i + 1:i + 2] in ('"', "\\")
|
||||
parsed.append(quoted[i + 1] if escaped else quoted[i])
|
||||
i += 2 if escaped else 1
|
||||
return "".join(parsed)
|
||||
if len(value) >= 2 and value[0] == value[-1] == "'":
|
||||
return value[1:-1]
|
||||
return value
|
||||
|
||||
|
||||
def load_env_file(env_path: Path) -> Dict[str, str]:
|
||||
"""Parse a ``.env`` file into a dict WITHOUT touching ``os.environ``: ``export``
|
||||
prefix, ``#`` comments, and the writer's quote escapes reversed via the canonical
|
||||
``_parse_env_value``. ``utf-8-sig`` so a BOM doesn't prefix the first key."""
|
||||
"""THE ``.env`` tokenizer: every reader (profile scope, ``hermes_cli.config.load_env``, the dashboard
|
||||
scrub, skill secret capture, managed .env, setup prompts) parses through here so no two boundaries
|
||||
disagree on which keys/values a file defines. Dict only — never touches ``os.environ``. ``export``
|
||||
prefix, ``#`` comments, quote escapes reversed; ``utf-8-sig`` so a BOM doesn't prefix the first key.
|
||||
Invalid UTF-8 decodes as latin-1, exactly like ``env_loader._load_dotenv_with_fallback`` installs it
|
||||
into ``os.environ``. Absent/unreadable → ``{}``."""
|
||||
secrets: Dict[str, str] = {}
|
||||
try:
|
||||
text = env_path.read_text(encoding="utf-8-sig")
|
||||
except (FileNotFoundError, OSError, UnicodeDecodeError):
|
||||
raw = env_path.read_bytes()
|
||||
except OSError:
|
||||
return secrets
|
||||
|
||||
from hermes_cli.config import _parse_env_value
|
||||
if raw.startswith(codecs.BOM_UTF8):
|
||||
raw = raw[len(codecs.BOM_UTF8):]
|
||||
try:
|
||||
text = raw.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
text = raw.decode("latin-1")
|
||||
|
||||
for raw in text.splitlines():
|
||||
line = raw.strip()
|
||||
|
||||
+3
-25
@@ -2352,24 +2352,6 @@ def save_config(
|
||||
_LAST_EXPANDED_CONFIG_BY_PATH[str(config_path)] = copy.deepcopy(current_normalized)
|
||||
|
||||
|
||||
def _parse_env_value(raw_value: str) -> str:
|
||||
"""Parse the small .env value subset Hermes writes itself (bare, 'single', or "double" with
|
||||
``\\"`` / ``\\\\`` escapes)."""
|
||||
value = raw_value.strip()
|
||||
if len(value) >= 2 and value[0] == value[-1] == '"':
|
||||
quoted = value[1:-1]
|
||||
parsed: list[str] = []
|
||||
i = 0
|
||||
while i < len(quoted):
|
||||
escaped = quoted[i] == "\\" and quoted[i + 1:i + 2] in ('"', "\\")
|
||||
parsed.append(quoted[i + 1] if escaped else quoted[i])
|
||||
i += 2 if escaped else 1
|
||||
return "".join(parsed)
|
||||
if len(value) >= 2 and value[0] == value[-1] == "'":
|
||||
return value[1:-1]
|
||||
return value
|
||||
|
||||
|
||||
# load_env() memo keyed on (path, mtime, size). Editing .env bumps mtime -> rebuild;
|
||||
# invalidate_env_cache() is the explicit knob for writers on coarse-mtime filesystems.
|
||||
_env_cache: Optional[Tuple[Tuple[str, Optional[float], Optional[int]], Dict[str, str]]] = None
|
||||
@@ -2391,13 +2373,9 @@ def load_env() -> Dict[str, str]:
|
||||
if cache_key is not None and _env_cache is not None and _env_cache[0] == cache_key:
|
||||
return dict(_env_cache[1])
|
||||
|
||||
env_vars: Dict[str, str] = {}
|
||||
for line in _read_env_lines(env_path) if env_path.exists() else ():
|
||||
line = line.strip()
|
||||
if line and not line.startswith('#') and '=' in line:
|
||||
# Bash-compatible ``export KEY=...`` parses as ``KEY``.
|
||||
key, _, value = line.removeprefix('export ').partition('=')
|
||||
env_vars[key.strip()] = _parse_env_value(value)
|
||||
from agent.secret_scope import load_env_file # the one .env tokenizer; also installs profile scopes
|
||||
|
||||
env_vars = load_env_file(env_path)
|
||||
if cache_key is not None:
|
||||
_env_cache = (cache_key, dict(env_vars))
|
||||
return env_vars
|
||||
|
||||
@@ -47,24 +47,11 @@ _PROFILE_MANAGED_ENV_KEYS: frozenset[str] = frozenset({
|
||||
|
||||
|
||||
def _env_keys_defined_in_dotenv(path: Path) -> set[str]:
|
||||
"""KEY names assigned in a dotenv file (including empty ``KEY=``). A fast line scanner (works in early
|
||||
bootstrap without python-dotenv); decode errors fall back to latin-1 like ``_load_dotenv_with_fallback``."""
|
||||
keys: set[str] = set()
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
try:
|
||||
text = path.read_text(encoding="latin-1", errors="replace")
|
||||
except Exception:
|
||||
return keys
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
key = line.removeprefix("export ").split("=", 1)[0].strip()
|
||||
if key:
|
||||
keys.add(key)
|
||||
return keys
|
||||
"""KEY names assigned in a dotenv file (including empty ``KEY=``), via the same tokenizer that installs
|
||||
profile scopes — a key the installer sees is a key the dashboard scrub sees (BOM'd first line included)."""
|
||||
from agent.secret_scope import load_env_file
|
||||
|
||||
return set(load_env_file(path))
|
||||
|
||||
|
||||
def _clear_known_keys_missing_from_dotenv(path: Path) -> None:
|
||||
|
||||
+10
-13
@@ -76,8 +76,7 @@ def _cached_read(path: Path, cache: Dict[str, tuple], parse):
|
||||
if hit is not None and hit[:2] == key:
|
||||
return copy.deepcopy(hit[2])
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
parsed = parse(f)
|
||||
parsed = parse(path)
|
||||
except Exception as exc: # noqa: BLE001 — fail-open, but LOUD
|
||||
logger.warning(
|
||||
"managed scope: failed to parse %s: %s — IGNORING this managed file. "
|
||||
@@ -99,12 +98,19 @@ def _load_managed_file(name: str, cache: Dict[str, tuple], parse) -> dict:
|
||||
|
||||
def load_managed_config() -> dict:
|
||||
"""Parsed managed config.yaml, or {} when absent/malformed (fail-open)."""
|
||||
return _load_managed_file("config.yaml", _CONFIG_CACHE, lambda f: yaml.safe_load(f) or {})
|
||||
return _load_managed_file("config.yaml", _CONFIG_CACHE, lambda p: yaml.safe_load(p.read_text(encoding="utf-8")) or {})
|
||||
|
||||
|
||||
def load_managed_env() -> Dict[str, str]:
|
||||
"""Parsed managed .env (KEY=VALUE), or {} when absent (fail-open)."""
|
||||
return _load_managed_file(".env", _ENV_CACHE, _parse_env)
|
||||
return _load_managed_file(".env", _ENV_CACHE, _parse_managed_env)
|
||||
|
||||
|
||||
def _parse_managed_env(path: Path) -> Dict[str, str]:
|
||||
from agent.secret_scope import load_env_file
|
||||
|
||||
path.read_text(encoding="utf-8-sig") # load_env_file swallows decode errors; an admin file must fail LOUD
|
||||
return load_env_file(path)
|
||||
|
||||
|
||||
def apply_managed_overlay(config: dict) -> dict:
|
||||
@@ -135,15 +141,6 @@ def apply_managed_overlay(config: dict) -> dict:
|
||||
return config
|
||||
|
||||
|
||||
def _parse_env(f) -> Dict[str, str]:
|
||||
out: Dict[str, str] = {}
|
||||
for line in map(str.strip, f):
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
key, _, value = line.partition("=")
|
||||
out[key.strip()] = value.strip().strip("\"'")
|
||||
return out
|
||||
|
||||
|
||||
def _flatten_keys(d: dict, prefix: str = "") -> set:
|
||||
keys: set = set()
|
||||
for k, v in d.items():
|
||||
|
||||
@@ -31,21 +31,10 @@ def _is_active(p, active: str) -> bool:
|
||||
|
||||
|
||||
def _env_file_has_key(env_path: Path, key: str) -> bool:
|
||||
"""True when *key* is assigned in *env_path*. Read as utf-8-sig: a Notepad-edited .env can
|
||||
carry a BOM that would hide the first key behind U+FEFF. A mis-encoded file (UnicodeDecodeError
|
||||
is a ValueError, not OSError) must not abort the install preview — skip the pre-check."""
|
||||
if not env_path.is_file():
|
||||
return False
|
||||
try:
|
||||
# .env is written as UTF-8 everywhere in the codebase, but a Notepad-edited file can carry a BOM —
|
||||
# read as utf-8-sig so the first key isn't hidden behind U+FEFF (#62617).
|
||||
for raw in env_path.read_text(encoding="utf-8-sig").splitlines():
|
||||
line = raw.strip()
|
||||
if line and not line.startswith("#") and line.split("=", 1)[0].strip() == key:
|
||||
return True
|
||||
except (OSError, UnicodeDecodeError):
|
||||
pass
|
||||
return False
|
||||
"""True when *key* is assigned in *env_path* (unreadable/mis-encoded file → False, never aborts)."""
|
||||
from agent.secret_scope import load_env_file
|
||||
|
||||
return key in load_env_file(env_path)
|
||||
|
||||
|
||||
def _render_distribution_plan(plan) -> None:
|
||||
|
||||
@@ -297,21 +297,10 @@ def _fire_cron_job_for_profile(profile: str, job_id: str, *, force: bool = False
|
||||
|
||||
|
||||
def _profile_env_value(home: Path, key: str) -> str:
|
||||
"""Best-effort read of one KEY=VALUE line from a profile's .env file."""
|
||||
try:
|
||||
env_path = home / ".env"
|
||||
if not env_path.is_file():
|
||||
return ""
|
||||
for line in env_path.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
k, v = line.split("=", 1)
|
||||
if k.strip() == key:
|
||||
return v.strip().strip('"').strip("'")
|
||||
except Exception:
|
||||
pass
|
||||
return ""
|
||||
"""One value from a profile's .env (``""`` when absent/unreadable)."""
|
||||
from agent.secret_scope import load_env_file
|
||||
|
||||
return load_env_file(home / ".env").get(key, "")
|
||||
|
||||
|
||||
def _gateway_fire_endpoint(profile: str, home: Path) -> str:
|
||||
|
||||
@@ -52,10 +52,10 @@ def _http_get(url: str, path: str, timeout: int):
|
||||
def _prompt_api_key(label: str, env_var: str, hermes_home: str) -> str:
|
||||
"""Prompt for API key, showing masked existing value if found."""
|
||||
existing = os.environ.get(env_var, "")
|
||||
env_path = Path(hermes_home) / ".env"
|
||||
if not existing and env_path.exists(): # utf-8-sig: a Notepad BOM on line 1 would otherwise defeat the key match
|
||||
lines = env_path.read_text(encoding="utf-8-sig", errors="replace").splitlines()
|
||||
existing = next((line.split("=", 1)[1].strip() for line in lines if line.startswith(f"{env_var}=")), "")
|
||||
if not existing:
|
||||
from agent.secret_scope import load_env_file
|
||||
|
||||
existing = load_env_file(Path(hermes_home) / ".env").get(env_var, "")
|
||||
hint = f" (current: {_masked(existing)}, blank to keep)" if existing else ""
|
||||
return getpass.getpass(f" {label} API key{hint}: ").strip()
|
||||
|
||||
|
||||
@@ -57,6 +57,36 @@ def test_utf8_bom_does_not_mangle_first_key(tmp_path, monkeypatch):
|
||||
assert os.environ.get("\ufeffFIRST_KEY") is None
|
||||
|
||||
|
||||
def test_bom_first_key_is_seen_by_installer_and_scrub_alike(tmp_path, monkeypatch):
|
||||
"""Invariant: the key set the dashboard/profile scrub computes (``_env_keys_defined_in_dotenv``) equals
|
||||
the key set the installers define (``load_hermes_dotenv`` into os.environ, ``load_env_file`` into a
|
||||
profile scope). A BOM'd first line, ``export``, quotes and inline comments must not split them —
|
||||
a key one side sees and the other doesn't is a scrub miss."""
|
||||
from hermes_cli.env_loader import _env_keys_defined_in_dotenv
|
||||
from agent.secret_scope import load_env_file
|
||||
|
||||
home = tmp_path / "hermes"
|
||||
home.mkdir()
|
||||
env_file = home / ".env"
|
||||
env_file.write_bytes(
|
||||
b"\xef\xbb\xbfFIRST_KEY=first-value\n"
|
||||
b"export EXPORTED_KEY='quoted # not a comment'\n"
|
||||
b"COMMENTED_KEY=value # trailing comment\n"
|
||||
b"EMPTY_KEY=\n"
|
||||
)
|
||||
for key in ("FIRST_KEY", "EXPORTED_KEY", "COMMENTED_KEY", "EMPTY_KEY", "\ufeffFIRST_KEY"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
|
||||
load_hermes_dotenv(hermes_home=home)
|
||||
installed = {k for k in ("FIRST_KEY", "EXPORTED_KEY", "COMMENTED_KEY", "EMPTY_KEY") if k in os.environ}
|
||||
scoped = load_env_file(env_file)
|
||||
|
||||
assert _env_keys_defined_in_dotenv(env_file) == installed == set(scoped)
|
||||
assert "\ufeffFIRST_KEY" not in _env_keys_defined_in_dotenv(env_file)
|
||||
assert scoped["EXPORTED_KEY"] == os.environ["EXPORTED_KEY"] == "quoted # not a comment"
|
||||
assert scoped["COMMENTED_KEY"] == os.environ["COMMENTED_KEY"] == "value"
|
||||
|
||||
|
||||
def test_bomless_utf8_env_still_loads(tmp_path, monkeypatch):
|
||||
"""BOM-less UTF-8 .env files must keep loading after utf-8-sig."""
|
||||
home = tmp_path / "hermes"
|
||||
|
||||
+5
-11
@@ -88,17 +88,11 @@ def _skill_lookup_path_error(name: str) -> Optional[str]:
|
||||
|
||||
|
||||
def load_env() -> Dict[str, str]:
|
||||
"""Load profile-scoped environment variables from HERMES_HOME/.env."""
|
||||
env_path = get_hermes_home() / ".env"
|
||||
env_vars: Dict[str, str] = {}
|
||||
if env_path.exists():
|
||||
# utf-8-sig: a Notepad BOM would otherwise glue U+FEFF onto the first key.
|
||||
with env_path.open(encoding="utf-8-sig", errors="replace") as f:
|
||||
for line in map(str.strip, f):
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
key, _, value = line.removeprefix("export ").partition("=")
|
||||
env_vars[key.strip()] = value.strip().strip("\"'")
|
||||
return env_vars
|
||||
"""Snapshot of HERMES_HOME/.env for the post-skill secret-capture diff (same tokenizer that
|
||||
installs the profile scope, so a captured value never differs from the served one)."""
|
||||
from agent.secret_scope import load_env_file
|
||||
|
||||
return load_env_file(get_hermes_home() / ".env")
|
||||
|
||||
|
||||
def set_secret_capture_callback(callback) -> None:
|
||||
|
||||
Reference in New Issue
Block a user