diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py index b4d44be88af2..dde61b43ecdd 100644 --- a/plugins/memory/openviking/__init__.py +++ b/plugins/memory/openviking/__init__.py @@ -836,6 +836,23 @@ def _precreate_secret_file(path: Path) -> None: logger.debug("Could not pre-create secret file %s: %s", path, e) +def _env_line_safe(value: Any) -> str: + """Neutralize characters that would break ``.env`` line structure. + + A ``.env`` file is strictly line-oriented (one ``KEY=VALUE`` per line), + and ``_write_env_vars`` interpolates each value straight into that line. + A value carrying an embedded CR/LF would therefore spill onto a new line + and be re-parsed as a *separate* ``KEY=VALUE`` entry on the next + ``read_text().splitlines()`` round-trip — letting a malformed or pasted + secret (e.g. an api_key copied with a trailing record) inject an + arbitrary additional variable into the persisted credentials file. Strip + the line terminators (``\r``, ``\n``) and the NUL byte so a value can + only ever occupy the single line it is written on. + """ + text = value if isinstance(value, str) else str(value) + return text.replace("\r", "").replace("\n", "").replace("\x00", "") + + def _write_env_vars(env_path: Path, env_writes: dict, remove_keys: tuple[str, ...] = ()) -> None: env_path.parent.mkdir(parents=True, exist_ok=True) remove_set = set(remove_keys) - set(env_writes) @@ -847,13 +864,13 @@ def _write_env_vars(env_path: Path, env_writes: dict, remove_keys: tuple[str, .. if key_match in remove_set: continue if key_match in env_writes: - new_lines.append(f"{key_match}={env_writes[key_match]}") + 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}={val}") + new_lines.append(f"{key}={_env_line_safe(val)}") # Pre-create with 0600 so secrets are never briefly world-readable. _precreate_secret_file(env_path) env_path.write_text("\n".join(new_lines) + ("\n" if new_lines else ""), encoding="utf-8") diff --git a/tests/plugins/memory/test_openviking_provider.py b/tests/plugins/memory/test_openviking_provider.py index 28f2d8e9d46a..46804f3b986a 100644 --- a/tests/plugins/memory/test_openviking_provider.py +++ b/tests/plugins/memory/test_openviking_provider.py @@ -89,6 +89,39 @@ def test_openviking_env_writer_restricts_file_permissions(tmp_path): assert stat.S_IMODE(env_path.stat().st_mode) == 0o600 +def test_openviking_env_writer_strips_embedded_newlines_in_values(tmp_path): + # A secret pasted with an embedded CR/LF must not spill onto a new line, + # or the round-trip re-parses the tail as a separate KEY=VALUE entry and + # injects an arbitrary variable into the persisted credentials file. + env_path = tmp_path / ".env" + + openviking_module._write_env_vars( + env_path, + {"OPENVIKING_API_KEY": "good\nINJECTED_KEY=attacker"}, + ) + + lines = env_path.read_text(encoding="utf-8").splitlines() + assert lines == ["OPENVIKING_API_KEY=goodINJECTED_KEY=attacker"] + # No injected line means a follow-up read sees no rogue key. + parsed = dict(line.split("=", 1) for line in lines if "=" in line) + assert set(parsed) == {"OPENVIKING_API_KEY"} + assert "INJECTED_KEY" not in parsed + + +def test_openviking_env_writer_strips_newlines_when_updating_existing_key(tmp_path): + env_path = tmp_path / ".env" + env_path.write_text("OPENVIKING_API_KEY=old\n", encoding="utf-8") + + openviking_module._write_env_vars( + env_path, + {"OPENVIKING_API_KEY": "new\r\nROGUE=1"}, + ) + + lines = env_path.read_text(encoding="utf-8").splitlines() + assert lines == ["OPENVIKING_API_KEY=newROGUE=1"] + assert all(not line.startswith("ROGUE=") for line in lines) + + @pytest.mark.skipif(os.name == "nt", reason="POSIX file modes") def test_ovcli_config_writer_restricts_file_permissions(tmp_path): config_path = tmp_path / "ovcli.conf"