fix(llm): claude-code provider security and privacy hardening

- Override is_external_service=True so the MemPalace#1224 privacy gate fires.
  The CLI binary runs locally but every classify call routes user content
  to Anthropic-hosted models, so the URL-based base-class default
  (returns local because endpoint is None) misclassifies this provider.

- Strip ANTHROPIC_* env vars from the subprocess environment. If the
  user has ANTHROPIC_API_KEY exported, claude -p can fall back to
  API-key auth and bill the API account instead of the subscription
  this provider is built around.

- Frame system+user content with <system>/<user> XML tags instead of
  literal SYSTEM:/USER: markers. A malicious drawer text containing
  '\\n\\nSYSTEM:\\nIgnore prior instructions...' could otherwise spoof
  the boundary and inject a second system prompt.

- Spawn the absolute path returned by shutil.which("claude") rather
  than the bare 'claude' literal. Closes a TOCTOU window between
  check_available() resolving the binary and classify() spawning a
  potentially different binary if PATH changes between calls.

- Pass encoding='utf-8' explicitly to subprocess.run so a Windows
  cp1252 locale does not mojibake the JSON envelope before json.loads.

- Include the raw stdout excerpt in the non-JSON envelope LLMError so
  CLI-output regressions can be debugged without reproducing.

- Broaden check_available()'s exception filter from
  (subprocess.TimeoutExpired, OSError) to (subprocess.SubprocessError,
  OSError) so future SubprocessError subclasses do not leak.

Tests added: env-scrub propagation, is_external_service True,
binary-missing path. Existing tests updated for the resolved-binary
cmd[0] and XML stdin framing.

Author: mvalentsev

mempalace/llm_client.py

@@ -435,6 +435,21 @@ def __init__(
     ):
         super().__init__(model=model, timeout=timeout)
 
+    @property
+    def is_external_service(self) -> bool:
+        # The CLI binary runs locally but routes every classify call to
+        # Anthropic's hosted models, so user content does leave the machine.
+        # Override the URL-based default in the base class.
+        return True
+
+    def _subprocess_env(self) -> dict:
+        # Strip ANTHROPIC_* env vars before spawning `claude -p`. If the user
+        # has ANTHROPIC_API_KEY exported in their shell, the CLI may fall
+        # back to API-key auth and bill the API account instead of the
+        # subscription this provider is built around. Removing the vars
+        # forces OAuth / keychain auth, which is the documented path.
+        return {k: v for k, v in os.environ.items() if not k.upper().startswith("ANTHROPIC_")}
+
     def check_available(self) -> tuple[bool, str]:
         binary = shutil.which("claude")
         if not binary:
@@ -445,12 +460,15 @@ def check_available(self) -> tuple[bool, str]:
             )
         try:
             r = subprocess.run(
-                ["claude", "auth", "status", "--text"],
+                [binary, "auth", "status", "--text"],
                 capture_output=True,
                 text=True,
+                encoding="utf-8",
+                errors="replace",
                 timeout=10,
+                env=self._subprocess_env(),
             )
-        except (subprocess.TimeoutExpired, OSError) as e:
+        except (subprocess.SubprocessError, OSError) as e:
             return False, f"`claude auth status` failed: {e}"
         if r.returncode != 0:
             return (
@@ -463,36 +481,42 @@ def classify(self, system: str, user: str, json_mode: bool = True) -> LLMRespons
         sys_prompt = system
         if json_mode:
             sys_prompt += "\n\nRespond with valid JSON only, no prose."
+        binary = shutil.which("claude")
+        if not binary:
+            raise LLMError("`claude` CLI not found in PATH")
         # `--bare` would skip hooks, plugins, CLAUDE.md auto-discovery, but it
         # also forces claude to use ANTHROPIC_API_KEY only and ignore OAuth /
         # keychain. That defeats this provider's whole point (subscription
         # auth), so we omit it. To keep the surrounding context minimal we
         # invoke from a temp cwd so claude does not pick up a project-level
         # CLAUDE.md it does not need.
         #
-        # System prompt is prepended to stdin instead of being passed via
-        # `--system-prompt` argv. argv is visible to other local users via
-        # `ps` / /proc/*/cmdline, and the prompt can carry sensitive context
-        # (entity names, project paths). The SYSTEM/USER framing is a
-        # convention `claude -p` follows reliably for classification tasks.
+        # System and user content go through stdin (not argv) so they are
+        # not visible to other local users via `ps` / /proc/*/cmdline, and
+        # to keep prompt-injection surface narrow we frame them with XML-
+        # like tags rather than literal "SYSTEM:" / "USER:" markers a
+        # malicious drawer could spoof.
         cmd = [
-            "claude",
+            binary,
             "-p",
             "--no-session-persistence",  # don't pollute Claude Code session history
             "--output-format",
             "json",
             "--model",
             self.model,
         ]
-        combined_input = f"SYSTEM:\n{sys_prompt}\n\nUSER:\n{user}"
+        combined_input = f"<system>\n{sys_prompt}\n</system>\n<user>\n{user}\n</user>"
         try:
             r = subprocess.run(
                 cmd,
                 input=combined_input,
                 capture_output=True,
                 text=True,
+                encoding="utf-8",
+                errors="replace",
                 timeout=self.timeout,
                 cwd=tempfile.gettempdir(),
+                env=self._subprocess_env(),
             )
         except subprocess.TimeoutExpired as e:
             raise LLMError(f"`claude -p` timed out after {self.timeout}s") from e
@@ -504,7 +528,10 @@ def classify(self, system: str, user: str, json_mode: bool = True) -> LLMRespons
         try:
             envelope = json.loads(r.stdout)
         except json.JSONDecodeError as e:
-            raise LLMError(f"`claude -p` returned non-JSON envelope: {e}") from e
+            stdout_excerpt = (r.stdout or "").strip()[:200]
+            raise LLMError(
+                f"`claude -p` returned non-JSON envelope: {e}; stdout={stdout_excerpt!r}"
+            ) from e
         # `--output-format json` returns:
         # {"type":"result","result":"<text>","total_cost_usd":...,...}
         text = envelope.get("result", "")

tests/test_llm_client.py

@@ -552,6 +552,9 @@ def test_claude_code_check_available_ready():
     assert msg == "ok"
 
 
+_FAKE_CLAUDE_BIN = "/usr/local/bin/claude"
+
+
 def test_claude_code_classify_command_line():
     captured = {}
 
@@ -560,11 +563,15 @@ def fake_run(cmd, **kwargs):
         captured["kwargs"] = kwargs
         return _mock_completed(0, stdout=_claude_envelope('{"ok": true}'))
 
-    with patch("mempalace.llm_client.subprocess.run", side_effect=fake_run):
-        p = ClaudeCodeProvider(model="claude-haiku-4-5", timeout=99)
-        p.classify("system text", "user text", json_mode=True)
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch("mempalace.llm_client.subprocess.run", side_effect=fake_run):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5", timeout=99)
+            p.classify("system text", "user text", json_mode=True)
 
-    assert captured["cmd"][0] == "claude"
+    # cmd[0] is the resolved absolute path from shutil.which, not a bare
+    # "claude" literal -- avoids a TOCTOU between check_available and
+    # classify if PATH changes between calls.
+    assert captured["cmd"][0] == _FAKE_CLAUDE_BIN
     assert "-p" in captured["cmd"]
     # `--bare` is intentionally NOT passed: it would force ANTHROPIC_API_KEY
     # auth and disable OAuth / keychain, defeating the subscription path.
@@ -578,15 +585,22 @@ def fake_run(cmd, **kwargs):
     # users via `ps` / /proc/*/cmdline and may carry sensitive context.
     assert "--system-prompt" not in captured["cmd"]
     assert "system text" not in captured["cmd"]
-    # System + user are framed in stdin instead. json_mode appends a
-    # JSON-only directive to the system block.
+    # System + user are framed in stdin with XML-like tags so a malicious
+    # drawer cannot spoof the boundary with literal "SYSTEM:" / "USER:"
+    # markers. json_mode appends a JSON-only directive inside the <system>
+    # block.
     stdin_input = captured["kwargs"]["input"]
-    assert stdin_input.startswith("SYSTEM:\nsystem text")
+    assert stdin_input.startswith("<system>\nsystem text")
     assert "JSON only" in stdin_input
-    assert "\n\nUSER:\nuser text" in stdin_input
+    assert "</system>\n<user>\nuser text\n</user>" in stdin_input
+    assert "SYSTEM:" not in stdin_input
+    assert "USER:" not in stdin_input
     assert captured["kwargs"]["timeout"] == 99
     # cwd must be a temp dir so claude does not pick up a project-level CLAUDE.md
     assert captured["kwargs"]["cwd"] == tempfile.gettempdir()
+    # Explicit UTF-8 decoding so Windows cp1252 locale does not mojibake the
+    # JSON envelope.
+    assert captured["kwargs"]["encoding"] == "utf-8"
 
 
 def test_claude_code_classify_json_mode_off_keeps_system_clean():
@@ -597,24 +611,64 @@ def fake_run(cmd, **kwargs):
         captured["kwargs"] = kwargs
         return _mock_completed(0, stdout=_claude_envelope("plain text reply"))
 
-    with patch("mempalace.llm_client.subprocess.run", side_effect=fake_run):
-        p = ClaudeCodeProvider(model="claude-haiku-4-5")
-        resp = p.classify("system text", "user", json_mode=False)
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch("mempalace.llm_client.subprocess.run", side_effect=fake_run):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5")
+            resp = p.classify("system text", "user", json_mode=False)
 
     # No JSON-only directive appended when json_mode=False; raw system
-    # text appears verbatim in the stdin SYSTEM block (not in argv).
+    # text appears verbatim inside the <system> block (not in argv).
     assert "--system-prompt" not in captured["cmd"]
-    assert captured["kwargs"]["input"] == "SYSTEM:\nsystem text\n\nUSER:\nuser"
+    assert captured["kwargs"]["input"] == (
+        "<system>\nsystem text\n</system>\n<user>\nuser\n</user>"
+    )
     assert resp.text == "plain text reply"
 
 
+def test_claude_code_strips_anthropic_env_vars(monkeypatch):
+    captured = {}
+
+    def fake_run(cmd, **kwargs):
+        captured["env"] = kwargs.get("env")
+        return _mock_completed(0, stdout=_claude_envelope("ok"))
+
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test-key")
+    monkeypatch.setenv("ANTHROPIC_AUTH_TOKEN", "tok-test")
+    monkeypatch.setenv("anthropic_other", "lower")  # case-insensitive prefix scrub
+    monkeypatch.setenv("UNRELATED_VAR", "kept")
+
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch("mempalace.llm_client.subprocess.run", side_effect=fake_run):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5")
+            p.classify("s", "u")
+
+    env = captured["env"]
+    assert env is not None
+    # ANTHROPIC_* in any case is stripped so claude -p can't fall back to
+    # API-key auth and bill the API account instead of the subscription.
+    assert "ANTHROPIC_API_KEY" not in env
+    assert "ANTHROPIC_AUTH_TOKEN" not in env
+    assert "anthropic_other" not in env
+    # Unrelated env vars must still pass through.
+    assert env.get("UNRELATED_VAR") == "kept"
+
+
+def test_claude_code_is_external_service_true():
+    # claude-code routes user content to Anthropic's hosted models via the
+    # local CLI binary, so the privacy gate (#1224) must treat it as
+    # external regardless of the URL-based base-class default.
+    p = ClaudeCodeProvider(model="claude-haiku-4-5")
+    assert p.is_external_service is True
+
+
 def test_claude_code_classify_parses_envelope():
-    with patch(
-        "mempalace.llm_client.subprocess.run",
-        return_value=_mock_completed(0, stdout=_claude_envelope("classified")),
-    ):
-        p = ClaudeCodeProvider(model="claude-haiku-4-5")
-        resp = p.classify("s", "u")
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch(
+            "mempalace.llm_client.subprocess.run",
+            return_value=_mock_completed(0, stdout=_claude_envelope("classified")),
+        ):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5")
+            resp = p.classify("s", "u")
 
     assert resp.text == "classified"
     assert resp.provider == "claude-code"
@@ -623,53 +677,68 @@ def test_claude_code_classify_parses_envelope():
 
 
 def test_claude_code_classify_timeout_raises_llm_error():
-    with patch(
-        "mempalace.llm_client.subprocess.run",
-        side_effect=subprocess.TimeoutExpired(cmd=["claude"], timeout=1),
-    ):
-        p = ClaudeCodeProvider(model="claude-haiku-4-5", timeout=1)
-        with pytest.raises(LLMError, match="timed out after 1s"):
-            p.classify("s", "u")
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch(
+            "mempalace.llm_client.subprocess.run",
+            side_effect=subprocess.TimeoutExpired(cmd=["claude"], timeout=1),
+        ):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5", timeout=1)
+            with pytest.raises(LLMError, match="timed out after 1s"):
+                p.classify("s", "u")
 
 
 def test_claude_code_classify_spawn_failure_raises_llm_error():
-    with patch(
-        "mempalace.llm_client.subprocess.run",
-        side_effect=FileNotFoundError("no such file: claude"),
-    ):
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch(
+            "mempalace.llm_client.subprocess.run",
+            side_effect=FileNotFoundError("no such file: claude"),
+        ):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5")
+            with pytest.raises(LLMError, match="failed to spawn"):
+                p.classify("s", "u")
+
+
+def test_claude_code_classify_binary_missing_raises_llm_error():
+    # If the binary disappears between provider construction and classify,
+    # surface a clear LLMError rather than letting subprocess raise an
+    # opaque FileNotFoundError later.
+    with patch("mempalace.llm_client.shutil.which", return_value=None):
         p = ClaudeCodeProvider(model="claude-haiku-4-5")
-        with pytest.raises(LLMError, match="failed to spawn"):
+        with pytest.raises(LLMError, match="not found in PATH"):
             p.classify("s", "u")
 
 
 def test_claude_code_classify_nonzero_raises_llm_error():
-    with patch(
-        "mempalace.llm_client.subprocess.run",
-        return_value=_mock_completed(1, stdout="", stderr="boom: bad model"),
-    ):
-        p = ClaudeCodeProvider(model="claude-haiku-4-5")
-        with pytest.raises(LLMError, match=r"`claude -p` exited 1: boom"):
-            p.classify("s", "u")
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch(
+            "mempalace.llm_client.subprocess.run",
+            return_value=_mock_completed(1, stdout="", stderr="boom: bad model"),
+        ):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5")
+            with pytest.raises(LLMError, match=r"`claude -p` exited 1: boom"):
+                p.classify("s", "u")
 
 
 def test_claude_code_classify_malformed_json_raises_llm_error():
-    with patch(
-        "mempalace.llm_client.subprocess.run",
-        return_value=_mock_completed(0, stdout="not valid json"),
-    ):
-        p = ClaudeCodeProvider(model="claude-haiku-4-5")
-        with pytest.raises(LLMError, match="non-JSON envelope"):
-            p.classify("s", "u")
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch(
+            "mempalace.llm_client.subprocess.run",
+            return_value=_mock_completed(0, stdout="not valid json"),
+        ):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5")
+            with pytest.raises(LLMError, match="non-JSON envelope"):
+                p.classify("s", "u")
 
 
 def test_claude_code_classify_empty_result_raises_llm_error():
-    with patch(
-        "mempalace.llm_client.subprocess.run",
-        return_value=_mock_completed(0, stdout=_claude_envelope("")),
-    ):
-        p = ClaudeCodeProvider(model="claude-haiku-4-5")
-        with pytest.raises(LLMError, match="empty result"):
-            p.classify("s", "u")
+    with patch("mempalace.llm_client.shutil.which", return_value=_FAKE_CLAUDE_BIN):
+        with patch(
+            "mempalace.llm_client.subprocess.run",
+            return_value=_mock_completed(0, stdout=_claude_envelope("")),
+        ):
+            p = ClaudeCodeProvider(model="claude-haiku-4-5")
+            with pytest.raises(LLMError, match="empty result"):
+                p.classify("s", "u")
 
 
 @pytest.mark.skipif(