fix(memory-setup): route .env writer through save_env_value's validation gate
hermes_cli/memory_setup.py::_write_env_vars() wrote provider-controlled .env entries with a direct Path.write_text() + post-hoc chmod, bypassing the denylist/regex/CRLF-stripping/atomic-replace validation that hermes_cli/config.py::save_env_value() already provides for every other .env writer in the codebase. A malicious or buggy memory-provider plugin declaring a crafted env-var name/value in its setup schema could inject arbitrary lines into .env. Routes memory-provider env writes through save_env_value(), and fixes a regression this surfaced in plugins/memory/supermemory/__init__.py:: post_setup(), which called the old two-parameter _write_env_vars(env_path, values) signature — restores the caller via context-local hermes_constants.set_hermes_home_override()/reset_hermes_home_override() instead of a removed env_path parameter, so explicit HERMES_HOME overrides during setup still resolve correctly. Adds test_env_file_created_with_secure_permissions, guarded on Windows (POSIX mode bits aren't enforced there, mirroring the existing skip in test_openviking_provider.py / test_supermemory_provider.py) since save_env_value's atomic-replace path creates the temp file at 0o600 before writing content, closing the TOCTOU window the old direct-write + chmod implementation had.
This commit is contained in:
pierrenode
+43
-43
@@ -11,7 +11,6 @@ import os
|
||||
import re
|
||||
import sys
|
||||
import shlex
|
||||
from pathlib import Path
|
||||
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.secret_prompt import masked_secret_prompt
|
||||
@@ -335,7 +334,6 @@ def cmd_setup(args) -> None:
|
||||
if not isinstance(provider_config, dict):
|
||||
provider_config = {}
|
||||
|
||||
env_path = get_hermes_home() / ".env"
|
||||
env_writes = {}
|
||||
|
||||
if schema:
|
||||
@@ -414,7 +412,7 @@ def cmd_setup(args) -> None:
|
||||
|
||||
# Write secrets to .env
|
||||
if env_writes:
|
||||
_write_env_vars(env_path, env_writes)
|
||||
_write_env_vars(env_writes)
|
||||
|
||||
print(f"\n Memory provider: {name}")
|
||||
print(" Activation saved to config.yaml")
|
||||
@@ -425,50 +423,52 @@ def cmd_setup(args) -> None:
|
||||
print("\n Start a new session to activate.\n")
|
||||
|
||||
|
||||
def _env_line_safe(value) -> str:
|
||||
"""Neutralize characters that would break ``.env`` line structure.
|
||||
def _write_env_vars(
|
||||
env_writes: dict,
|
||||
hermes_home: str | os.PathLike[str] | None = None,
|
||||
) -> None:
|
||||
"""Persist memory-provider env vars through the canonical ``.env`` writer.
|
||||
|
||||
``.env`` is strictly line-oriented (one ``KEY=VALUE`` per line) and
|
||||
values are interpolated straight into that line. A pasted secret with an
|
||||
embedded CR/LF would spill onto a new line and be re-parsed as a
|
||||
*separate* ``KEY=VALUE`` entry on the next read — injecting an arbitrary
|
||||
variable into the credentials file. Strip every separator recognized by
|
||||
``str.splitlines()`` plus NUL so a value can only occupy its own line.
|
||||
Mirrors the openviking plugin's writer and ``config.save_env_value``.
|
||||
Delegates to ``hermes_cli.config.save_env_value`` so every key flows
|
||||
through the same input-validation gate as every other ``.env`` writer:
|
||||
the ``_ENV_VAR_NAME_RE`` regex (no malformed identifiers), the
|
||||
``_ENV_VAR_NAME_DENYLIST`` (no ``LD_PRELOAD`` / ``PYTHONPATH`` /
|
||||
``HERMES_HOME`` / etc.), CR/LF stripping on the value, and the atomic
|
||||
0o600-from-creation write (no TOCTOU permission window). This function
|
||||
previously wrote via ``Path.write_text`` directly, bypassing all of
|
||||
that: a memory-provider plugin schema declaring ``env_var: "LD_PRELOAD"``
|
||||
would land in ``.env`` verbatim and load via the ``env_loader.py``
|
||||
``.env`` -> ``os.environ`` chain on the next Hermes startup, and the
|
||||
file existed at the default umask between the write and the later
|
||||
``chmod`` regardless of key legitimacy.
|
||||
|
||||
Validation failures (``ValueError`` from ``save_env_value`` — a
|
||||
denylisted name or an identifier rejected by ``_ENV_VAR_NAME_RE``) are
|
||||
surfaced and skipped rather than aborting the wizard, so a single bad
|
||||
key from one schema field doesn't take down the rest of the batch.
|
||||
Non-validation errors (filesystem failures, permission errors) are
|
||||
intentionally NOT caught — those indicate the wizard cannot safely
|
||||
persist any subsequent key either and should propagate.
|
||||
|
||||
``hermes_home`` may be supplied by plugin ``post_setup`` hooks that
|
||||
already received an explicit home directory (e.g. a non-default
|
||||
profile). It is applied through the context-local Hermes home override
|
||||
so ``save_env_value`` still owns the validation, sanitization, and
|
||||
atomic-write path without mutating global ``os.environ``.
|
||||
"""
|
||||
text = value if isinstance(value, str) else str(value)
|
||||
return "".join(text.replace("\x00", "").splitlines())
|
||||
from hermes_cli.config import save_env_value
|
||||
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
|
||||
|
||||
|
||||
def _write_env_vars(env_path: Path, env_writes: dict) -> None:
|
||||
"""Append or update env vars in .env file."""
|
||||
env_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
existing_lines = []
|
||||
if env_path.exists():
|
||||
existing_lines = env_path.read_text(encoding="utf-8").splitlines()
|
||||
|
||||
updated_keys = set()
|
||||
new_lines = []
|
||||
for line in existing_lines:
|
||||
key_match = line.split("=", 1)[0].strip() if "=" in line else ""
|
||||
if key_match in env_writes:
|
||||
new_lines.append(f"{key_match}={_env_line_safe(env_writes[key_match])}")
|
||||
updated_keys.add(key_match)
|
||||
else:
|
||||
new_lines.append(line)
|
||||
|
||||
for key, val in env_writes.items():
|
||||
if key not in updated_keys:
|
||||
new_lines.append(f"{key}={_env_line_safe(val)}")
|
||||
|
||||
env_path.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
|
||||
# Restrict permissions — .env holds API keys and tokens.
|
||||
token = set_hermes_home_override(hermes_home) if hermes_home is not None else None
|
||||
try:
|
||||
import stat
|
||||
env_path.chmod(stat.S_IRUSR | stat.S_IWUSR) # 0600
|
||||
except OSError:
|
||||
pass # Windows or read-only FS
|
||||
for key, val in env_writes.items():
|
||||
try:
|
||||
save_env_value(key, val)
|
||||
except ValueError as exc:
|
||||
print(f" Skipping {key}: {exc}")
|
||||
finally:
|
||||
if token is not None:
|
||||
reset_hermes_home_override(token)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -601,8 +601,6 @@ class SupermemoryMemoryProvider(MemoryProvider):
|
||||
return {"summary": _format_connection_summary(status)}
|
||||
|
||||
def post_setup(self, hermes_home: str, config: dict) -> None:
|
||||
from pathlib import Path
|
||||
|
||||
from hermes_cli.config import save_config
|
||||
from hermes_cli.memory_setup import _prompt, _write_env_vars
|
||||
|
||||
@@ -625,7 +623,7 @@ class SupermemoryMemoryProvider(MemoryProvider):
|
||||
save_config(config)
|
||||
|
||||
if env_writes:
|
||||
_write_env_vars(Path(hermes_home) / ".env", env_writes)
|
||||
_write_env_vars(env_writes, hermes_home=hermes_home)
|
||||
|
||||
api_key = env_writes.get("SUPERMEMORY_API_KEY") or existing
|
||||
# Make the freshly-entered key visible to the connection probe below.
|
||||
|
||||
@@ -44,22 +44,12 @@ def test_cmd_setup_generic_choice_cancel_writes_nothing(tmp_path, monkeypatch):
|
||||
assert not (tmp_path / ".env").exists()
|
||||
|
||||
|
||||
def test_write_env_vars_strips_line_separators_and_nul(tmp_path):
|
||||
"""A pasted secret with embedded CR/LF/NUL must not inject an extra
|
||||
KEY=VALUE line into .env (mirrors the openviking plugin's writer)."""
|
||||
env_path = tmp_path / ".env"
|
||||
|
||||
memory_setup._write_env_vars(
|
||||
env_path,
|
||||
{"PROVIDER_API_KEY": "good\nINJECTED_KEY=attacker\r\u2028\x00tail"},
|
||||
)
|
||||
|
||||
lines = env_path.read_text(encoding="utf-8").splitlines()
|
||||
assert lines == ["PROVIDER_API_KEY=goodINJECTED_KEY=attackertail"]
|
||||
parsed = dict(line.split("=", 1) for line in lines if "=" in line)
|
||||
assert set(parsed) == {"PROVIDER_API_KEY"}
|
||||
|
||||
|
||||
# _write_env_vars's CR/LF-stripping, denylist, and plain-value-roundtrip
|
||||
# behavior is covered by tests/hermes_cli/test_memory_setup_env_denylist.py,
|
||||
# which exercises the current save_env_value-routed signature
|
||||
# (env_writes, hermes_home=None) \u2014 these three tests pinned the prior direct
|
||||
# Path.write_text(env_path, env_writes) signature/implementation and were
|
||||
# removed along with it (#60587).
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
"""Tests for the env-write denylist on the memory-setup ``.env`` writer.
|
||||
|
||||
``hermes_cli.memory_setup._write_env_vars`` persists provider plugin
|
||||
credentials to ``~/.hermes/.env``. It previously called ``Path.write_text``
|
||||
directly, bypassing the ``_ENV_VAR_NAME_DENYLIST`` / ``_ENV_VAR_NAME_RE`` /
|
||||
CR-LF-stripping gates that ``save_env_value`` enforces for every other
|
||||
``.env`` writer in the codebase, and left the file at the default umask
|
||||
between the write and a later ``chmod`` (a TOCTOU permission window).
|
||||
|
||||
A memory provider plugin schema declaring ``env_var: "LD_PRELOAD"`` (or any
|
||||
other subprocess-influencing or Hermes-runtime-location name) could
|
||||
otherwise plant a value into ``.env`` via the interactive memory-setup
|
||||
wizard. The next Hermes process would load it through the
|
||||
``env_loader.py`` ``.env -> os.environ`` chain and execute attacker code
|
||||
before ``main()``.
|
||||
|
||||
The fix routes through ``save_env_value`` so the same gates fire.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.config import ensure_hermes_home, get_env_path, load_env
|
||||
from hermes_cli.memory_setup import _write_env_vars
|
||||
|
||||
|
||||
def _env_file_keys() -> set[str]:
|
||||
"""Parse ``~/.hermes/.env`` directly and return the set of keys present.
|
||||
|
||||
Used by tests that want to verify a key was NOT written to disk without
|
||||
going through ``load_env()`` (whose sanitization/caching could mask the
|
||||
underlying file state).
|
||||
"""
|
||||
env_path = get_env_path()
|
||||
if not env_path.exists():
|
||||
return set()
|
||||
keys: set[str] = set()
|
||||
for line in env_path.read_text(encoding="utf-8-sig").splitlines():
|
||||
line = line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
key, _, _ = line.partition("=")
|
||||
keys.add(key.strip())
|
||||
return keys
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _hermes_home(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
ensure_hermes_home()
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"denied_key",
|
||||
[
|
||||
"LD_PRELOAD",
|
||||
"LD_LIBRARY_PATH",
|
||||
"DYLD_INSERT_LIBRARIES",
|
||||
"PYTHONPATH",
|
||||
"PYTHONHOME",
|
||||
"NODE_OPTIONS",
|
||||
"PATH",
|
||||
"EDITOR",
|
||||
"GIT_SSH_COMMAND",
|
||||
"HERMES_HOME",
|
||||
"HERMES_PROFILE",
|
||||
"HERMES_CONFIG",
|
||||
"HERMES_ENV",
|
||||
],
|
||||
)
|
||||
def test_denylisted_key_is_skipped(denied_key, capsys):
|
||||
"""Each denylisted name must not land in .env even though the
|
||||
memory-setup wizard accepted it from a (hypothetically malicious)
|
||||
provider schema. The wizard prints a warning and continues."""
|
||||
_write_env_vars({denied_key: "/tmp/evil.so"})
|
||||
|
||||
# Assert directly against ``.env`` file contents so the test isn't
|
||||
# coupled to ``load_env()``'s sanitization or any future merge with
|
||||
# ``os.environ`` (where common names like ``PATH``/``EDITOR`` would
|
||||
# always appear and trivially satisfy a ``not in env`` check).
|
||||
assert denied_key not in _env_file_keys()
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert denied_key in captured.out
|
||||
assert "denylist" in captured.out.lower() or "Skipping" in captured.out
|
||||
|
||||
|
||||
def test_denylisted_key_does_not_block_other_writes(capsys):
|
||||
"""If a single batch contains one denylisted key plus legitimate
|
||||
integration credentials, the denylisted one is skipped but the
|
||||
legitimate ones still land. The wizard must not abort mid-batch."""
|
||||
_write_env_vars({
|
||||
"LD_PRELOAD": "/tmp/evil.so",
|
||||
"HERMES_LANGFUSE_PUBLIC_KEY": "pk-test-123",
|
||||
"OPENROUTER_API_KEY": "sk-or-test-456",
|
||||
})
|
||||
|
||||
assert "LD_PRELOAD" not in _env_file_keys()
|
||||
env = load_env()
|
||||
assert env["HERMES_LANGFUSE_PUBLIC_KEY"] == "pk-test-123"
|
||||
assert env["OPENROUTER_API_KEY"] == "sk-or-test-456"
|
||||
|
||||
|
||||
def test_legitimate_hermes_integration_key_still_writable():
|
||||
"""``HERMES_*`` overall is NOT blocked — only the four runtime
|
||||
location names (HOME/PROFILE/CONFIG/ENV). Integration credentials
|
||||
following the ``HERMES_*`` convention (HERMES_LANGFUSE_*,
|
||||
HERMES_SPOTIFY_*, HERMES_QWEN_BASE_URL, ...) must keep working or
|
||||
the memory-setup wizard regresses for every plugin that follows
|
||||
the convention."""
|
||||
_write_env_vars({
|
||||
"HERMES_LANGFUSE_PUBLIC_KEY": "pk-lf-789",
|
||||
"HERMES_QWEN_BASE_URL": "https://example.com/v1",
|
||||
})
|
||||
|
||||
env = load_env()
|
||||
assert env["HERMES_LANGFUSE_PUBLIC_KEY"] == "pk-lf-789"
|
||||
assert env["HERMES_QWEN_BASE_URL"] == "https://example.com/v1"
|
||||
|
||||
|
||||
def test_malformed_key_name_is_skipped(capsys):
|
||||
"""The canonical writer also enforces ``_ENV_VAR_NAME_RE`` —
|
||||
identifiers must match ``[A-Za-z_][A-Za-z0-9_]*``. A plugin schema
|
||||
declaring ``env_var: "FOO BAR"`` (space) was previously persisted
|
||||
verbatim, producing a malformed ``.env`` line."""
|
||||
_write_env_vars({"FOO BAR": "value"})
|
||||
|
||||
keys = _env_file_keys()
|
||||
assert "FOO BAR" not in keys
|
||||
assert "FOO" not in keys # not silently truncated either
|
||||
|
||||
captured = capsys.readouterr()
|
||||
assert "Skipping" in captured.out or "FOO BAR" in captured.out
|
||||
|
||||
|
||||
def test_legitimate_value_writes_round_trip():
|
||||
"""Negative control — the gate must not regress on a normal write."""
|
||||
_write_env_vars({"MEM0_API_KEY": "m0-test-key-abc"})
|
||||
|
||||
env = load_env()
|
||||
assert env["MEM0_API_KEY"] == "m0-test-key-abc"
|
||||
|
||||
|
||||
def test_explicit_hermes_home_writes_to_that_env_file(tmp_path):
|
||||
"""Plugin ``post_setup`` hooks (e.g. Supermemory) pass an explicit
|
||||
Hermes home; keep that target while still routing through
|
||||
``save_env_value`` validation instead of writing directly."""
|
||||
home = tmp_path / "plugin-home"
|
||||
|
||||
_write_env_vars({"MEM0_API_KEY": "m0-test-key-abc"}, hermes_home=home)
|
||||
|
||||
assert "MEM0_API_KEY=m0-test-key-abc\n" in (home / ".env").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def test_value_with_embedded_newline_is_stripped():
|
||||
"""``save_env_value`` strips CR/LF from the value to prevent
|
||||
.env-file structure injection (a value containing ``\\n`` would
|
||||
otherwise split the line and inject an arbitrary follow-on key).
|
||||
Routing through it gives the memory-setup wizard the same
|
||||
protection."""
|
||||
_write_env_vars({"MEM0_API_KEY": "key1\nEVIL=injected\n"})
|
||||
|
||||
env = load_env()
|
||||
# CR/LF stripped, value still lands intact (minus the newlines)
|
||||
assert env["MEM0_API_KEY"] == "key1EVIL=injected"
|
||||
# And no smuggled key landed — assert against the file too so the test
|
||||
# holds even if ``load_env()`` ever starts merging ``os.environ``.
|
||||
assert "EVIL" not in _env_file_keys()
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits not enforced on Windows")
|
||||
def test_env_file_created_with_secure_permissions(tmp_path):
|
||||
"""Regression guard for the TOCTOU window the direct ``Path.write_text``
|
||||
+ post-hoc ``chmod`` implementation had: ``save_env_value`` creates the
|
||||
temp file with 0o600 before any content is written and atomically
|
||||
replaces the target, so the file is never briefly world/group-readable
|
||||
at the process umask."""
|
||||
import stat
|
||||
|
||||
_write_env_vars({"MEM0_API_KEY": "m0-test-key-abc"})
|
||||
|
||||
env_path = get_env_path()
|
||||
mode = stat.S_IMODE(env_path.stat().st_mode)
|
||||
assert mode == stat.S_IRUSR | stat.S_IWUSR
|
||||
Reference in New Issue
Block a user