diff --git a/agent/secret_scope.py b/agent/secret_scope.py index 7b84dcbdc4..da5d9b927f 100644 --- a/agent/secret_scope.py +++ b/agent/secret_scope.py @@ -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() diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 4669b49e8e..ad60058c4c 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -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 diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index 4b7e83bd2c..f1c5c48f2a 100644 --- a/hermes_cli/env_loader.py +++ b/hermes_cli/env_loader.py @@ -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: diff --git a/hermes_cli/managed_scope.py b/hermes_cli/managed_scope.py index dc03b117c9..ef954fb86a 100644 --- a/hermes_cli/managed_scope.py +++ b/hermes_cli/managed_scope.py @@ -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(): diff --git a/hermes_cli/profile_cmd.py b/hermes_cli/profile_cmd.py index 45b67323a9..caf12315ea 100644 --- a/hermes_cli/profile_cmd.py +++ b/hermes_cli/profile_cmd.py @@ -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: diff --git a/hermes_cli/web_server_cron.py b/hermes_cli/web_server_cron.py index e894a23494..63e301a67b 100644 --- a/hermes_cli/web_server_cron.py +++ b/hermes_cli/web_server_cron.py @@ -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: diff --git a/plugins/memory/mem0/_setup.py b/plugins/memory/mem0/_setup.py index b623baa9c8..60bcbfc3ce 100644 --- a/plugins/memory/mem0/_setup.py +++ b/plugins/memory/mem0/_setup.py @@ -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() diff --git a/tests/hermes_cli/test_env_loader.py b/tests/hermes_cli/test_env_loader.py index 1eb34a41c0..4ab389ffe4 100644 --- a/tests/hermes_cli/test_env_loader.py +++ b/tests/hermes_cli/test_env_loader.py @@ -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" diff --git a/tools/skills_tool.py b/tools/skills_tool.py index 63e4ff1d17..3e50e1ec7a 100644 --- a/tools/skills_tool.py +++ b/tools/skills_tool.py @@ -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: