fix(cli): pass API key via env var instead of argv (closes #4862)

cli/python/src/mem0_cli/telemetry.py

@@ -4,6 +4,9 @@
 (telemetry_sender.py). The parent CLI process exits immediately; the
 subprocess handles email resolution, caching, and the HTTP POST.
 
+The Mem0 API key is passed via environment variable (not command-line argument)
+to avoid exposure through /proc/<pid>/cmdline, ps, and process inspection tools.
+
 Disable with: MEM0_TELEMETRY=false
 """
 
@@ -129,18 +132,25 @@ def capture_event(
             "payload": payload,
             "posthog_host": POSTHOG_HOST,
             "needs_email": not distinct_id or "@" not in distinct_id,
-            "mem0_api_key": config.platform.api_key or "",
             "mem0_base_url": config.platform.base_url or "https://api.mem0.ai",
             "config_path": str(CONFIG_FILE),
             "anon_distinct_id_to_alias": anon_id_to_alias,
         }
 
+        # Pass API key via environment variable instead of argv to prevent
+        # exposure through /proc/<pid>/cmdline, ps, and process inspectors.
+        # subprocess.Popen inherits the full env dict automatically.
+        env = {**os.environ}
+        if config.platform.api_key:
+            env["MEM0_API_KEY"] = config.platform.api_key
+
         subprocess.Popen(
             [sys.executable, "-m", "mem0_cli.telemetry_sender", json.dumps(context)],
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
             start_new_session=True,
             close_fds=True,
+            env=env,
         )
     except Exception:
         pass

cli/python/src/mem0_cli/telemetry_sender.py

@@ -5,9 +5,11 @@
 This module is spawned by telemetry.capture_event() and runs independently
 of the parent CLI process. It:
 
-1. Resolves the user's email via /v1/ping/ if not already cached
-2. Caches the email in ~/.mem0/config.json for future runs
-3. Sends the PostHog event
+1. Reads the Mem0 API key from the MEM0_API_KEY environment variable
+   (not from the command-line, to avoid exposing it via ps/cmdline).
+2. Resolves the user's email via /v1/ping/ if not already cached.
+3. Caches the email in ~/.mem0/config.json for future runs.
+4. Sends the PostHog event.
 
 All errors are silently swallowed — this process must never produce output
 or affect the user experience.
@@ -16,6 +18,7 @@
 from __future__ import annotations
 
 import json
+import os
 import sys
 import urllib.request
 
@@ -24,8 +27,11 @@ def main() -> None:
     ctx = json.loads(sys.argv[1])
     payload = ctx["payload"]
 
-    if ctx.get("needs_email") and ctx.get("mem0_api_key"):
-        _resolve_and_cache_email(ctx, payload)
+    # Read API key from environment variable (not argv) to avoid exposure.
+    mem0_api_key = os.environ.get("MEM0_API_KEY", "")
+
+    if ctx.get("needs_email") and mem0_api_key:
+        _resolve_and_cache_email(ctx, payload, mem0_api_key)
 
     # Fire $identify *after* email resolution so PostHog links the stored
     # anonymous id directly to the final identity (email, not the api-key
@@ -52,14 +58,14 @@ def _send_identify_event(ctx: dict, payload: dict, anon_id: str) -> None:
     _send_posthog_event(ctx["posthog_host"], identify_payload)
 
 
-def _resolve_and_cache_email(ctx: dict, payload: dict) -> None:
+def _resolve_and_cache_email(ctx: dict, payload: dict, mem0_api_key: str) -> None:
     """Call /v1/ping/ to get the user's email, update the payload, and cache it."""
     try:
         ping_url = ctx["mem0_base_url"].rstrip("/") + "/v1/ping/"
         req = urllib.request.Request(
             ping_url,
             headers={
-                "Authorization": "Token " + ctx["mem0_api_key"],
+                "Authorization": "Token " + mem0_api_key,
                 "Content-Type": "application/json",
             },
         )

cli/python/tests/test_telemetry_security.py

@@ -0,0 +1,120 @@
+"""Tests for CLI telemetry — focused on security and env-var API key handling."""
+
+from __future__ import annotations
+
+import json
+import os
+from unittest.mock import patch
+
+import pytest
+
+from mem0_cli.telemetry import capture_event
+
+
+class TestTelemetrySecurity:
+    """Security regression tests for telemetry subprocess spawning.
+
+    Regression: API keys were previously passed via sys.argv (visible in ps/cmdline).
+    Fix: API key is now passed via MEM0_API_KEY environment variable.
+    See: https://github.com/mem0ai/mem0/issues/4862
+    """
+
+    @patch("mem0_cli.telemetry.subprocess.Popen")
+    @patch("mem0_cli.telemetry.load_config")
+    @patch("mem0_cli.telemetry.is_agent_mode")
+    @patch("mem0_cli.telemetry._get_distinct_id")
+    def test_api_key_not_in_subprocess_argv(
+        self, mock_get_distinct, mock_agent_mode, mock_load_config, mock_popen
+    ) -> None:
+        """Verify the API key is never included in the subprocess command-line args."""
+        mock_get_distinct.return_value = "test@example.com"
+        mock_agent_mode.return_value = False
+        mock_config = _make_config(api_key="sk-verysecret123")
+        mock_load_config.return_value = mock_config
+
+        capture_event("test_event", properties={"foo": "bar"})
+
+        mock_popen.assert_called_once()
+        call_args = mock_popen.call_args
+
+        argv_list = call_args[0][0]  # positional args to Popen
+        # The JSON context is the last argv element
+        context_json = argv_list[-1]
+        context = json.loads(context_json)
+
+        # API key must NOT be in the context dict (was previously in mem0_api_key field)
+        assert "mem0_api_key" not in context, "API key leaked into subprocess argv"
+        assert "sk-verysecret123" not in context_json, "API key leaked into subprocess argv"
+
+    @patch("mem0_cli.telemetry.subprocess.Popen")
+    @patch("mem0_cli.telemetry.load_config")
+    @patch("mem0_cli.telemetry.is_agent_mode")
+    @patch("mem0_cli.telemetry._get_distinct_id")
+    def test_api_key_passed_via_env_var(
+        self, mock_get_distinct, mock_agent_mode, mock_load_config, mock_popen
+    ) -> None:
+        """Verify the API key is passed via MEM0_API_KEY environment variable."""
+        mock_get_distinct.return_value = "test@example.com"
+        mock_agent_mode.return_value = False
+        mock_config = _make_config(api_key="sk-testkey456")
+        mock_load_config.return_value = mock_config
+
+        capture_event("test_event")
+
+        mock_popen.assert_called_once()
+        call_kwargs = call_args_from_mock(call_args=mock_popen.call_args)
+
+        env = call_kwargs.get("env") or os.environ
+        assert "MEM0_API_KEY" in env, "MEM0_API_KEY not set in subprocess env"
+        assert env["MEM0_API_KEY"] == "sk-testkey456"
+
+    @patch("mem0_cli.telemetry.subprocess.Popen")
+    @patch("mem0_cli.telemetry.load_config")
+    @patch("mem0_cli.telemetry.is_agent_mode")
+    @patch("mem0_cli.telemetry._get_distinct_id")
+    def test_no_api_key_means_no_env_var(
+        self, mock_get_distinct, mock_agent_mode, mock_load_config, mock_popen
+    ) -> None:
+        """When no API key is configured, MEM0_API_KEY should not be set."""
+        mock_get_distinct.return_value = "anon-user"
+        mock_agent_mode.return_value = False
+        mock_config = _make_config(api_key="")
+        mock_load_config.return_value = mock_config
+
+        capture_event("test_event")
+
+        mock_popen.assert_called_once()
+        call_kwargs = call_args_from_mock(call_args=mock_popen.call_args)
+        env = call_kwargs.get("env")
+        if env is not None:
+            assert "MEM0_API_KEY" not in env
+
+
+# ── Helpers ────────────────────────────────────────────────────────────────────
+
+
+def call_args_from_mock(call_args) -> dict:
+    """Extract kwargs from a unittest.mock call args object."""
+    # call_args[1] is kwargs dict
+    return call_args[1] if len(call_args) > 1 else {}
+
+
+def _make_config(api_key: str = "", base_url: str = "https://api.mem0.ai"):
+    """Return a minimal mock config object matching the real config structure."""
+
+    class PlatformCfg:
+        def __init__(self):
+            self.api_key = api_key
+            self.base_url = base_url
+            self.user_email = ""
+
+    class TelemetryCfg:
+        def __init__(self):
+            self.anonymous_id = "test-anon-123"
+
+    class Config:
+        def __init__(self):
+            self.platform = PlatformCfg()
+            self.telemetry = TelemetryCfg()
+
+    return Config()