fix(code-mode): close proxy-laundering, ..-traversal, and 5 other Patch76 review findings

Implements the 9-item plan in
#854 (comment)
addressing the CHANGES_REQUESTED review at
#854 (review)
on commit 11ba402.

Blocker 1 — recursive-self-call guard bypassable via ha_call_write_tool
-----------------------------------------------------------------------

Two-layer fix:

* ``CategorizedSearchTransform`` gains an ``enable_code_mode: bool``
  constructor parameter (default False, preserves prior behaviour for
  installations that aren't running code mode). When True,
  ``_rebuild_category_cache`` swaps ``get_tool_catalog(ctx)`` for
  ``_get_visible_tools(ctx)`` — the same FastMCP helper that
  ``BM25SearchTransform`` already uses. Pinned tools (including
  ``ha_manage_custom_tool``) drop out of ``_read_tools`` /
  ``_write_tools`` / ``_delete_tools``, so ``categorized_call`` for a
  pinned name falls through to the ``RESOURCE_NOT_FOUND`` branch
  rather than dispatching the underlying tool. ``server.py`` flips the
  flag on whenever ``settings.enable_code_mode`` is True.

* ``_BLOCKED_TOOLS`` (sandbox-side defense in depth) now also includes
  the four search-transform synthetics: ``ha_search_tools``,
  ``ha_call_read_tool``, ``ha_call_write_tool``, ``ha_call_delete_tool``.
  Even if a future regression re-enables the proxy dispatch, sandbox
  code can't reach the laundering path. Direct calls to underlying
  tools by their real name (``call_tool("ha_get_history", ...)``) keep
  working — the block is on the synthetics only, so individual
  underlying tools can still be denylisted in the future without
  needing to rework the proxy.

Blocker 2 — ``..`` traversal in ``_normalize_endpoint``
-------------------------------------------------------

httpx happily resolves ``base_url='http://ha:8123/api'`` +
``../auth/providers`` to ``http://ha:8123/auth/providers``, escaping
the ``/api/`` prefix. After ``lstrip("/")`` and the optional ``api/``
strip, ``_normalize_endpoint`` now splits on ``/`` and rejects any
segment exactly equal to ``..``. ``..bar`` (filename starting with
two dots) and ``...`` (three dots) stay allowed — they aren't
traversal segments, just unusual filenames.

Verified against 12 endpoint cases covering all 4 traversal patterns
plus prior security guards (absolute URL, protocol-relative,
userinfo) still holding.

Additional registry blocks (``_BLOCKED_WS_COMMANDS``)
-----------------------------------------------------

Added 9 entries for floor / label / category registry mutations to
match the existing area / device / entity coverage. Each has a
wrapping MCP tool (``ha_config_set_floor`` / ``ha_config_set_label`` /
``ha_config_set_category``) so the same "force through the validated
path" rationale applies.

Additional event blocks (``_BLOCKED_HA_INTERNAL_EVENTS``)
---------------------------------------------------------

Added ``script_finished`` (pairs with the existing ``script_started``
block) and ``logbook_entry`` (this event IS the documented logbook
write API; spoofing injects fabricated rows directly into the user's
primary investigation tool — data-integrity issue, not just attack
surface). ``automation_triggered`` and ``call_service`` stay allowed
— legit "verify my handler reacts" use cases and downstream
consumers can already check event context for provenance.

``list_saved`` shape — nest under ``data.saved_tools``
------------------------------------------------------

``_SAVE_NAME_PATTERN`` accepts every key the *other* response shapes
use (``result``, ``code``, ``justification``, ``saved_tool``, ``count``).
A consumer doing ``r["data"]["result"]`` after a list_saved call
would have gotten a saved-tool entry instead of a run-result. Fixed
by nesting the dict under a stable ``saved_tools`` key. Updated
``test_list_saved_tools`` to pin the new shape.

Log injection (``%s`` interpolation of LLM-controlled strings)
--------------------------------------------------------------

Added ``_log_safe`` helper that replaces ``\r`` / ``\n`` / ``\t``
with spaces and truncates to 200 chars. Applied to the
``ha_manage_custom_tool invoked — justification: %s`` log line at
``tools_code.py:1151`` so an LLM-supplied
``"real reason\nFAKE_CRITICAL: …"`` cannot inject a synthetic second
log line. Audit-log endpoint/type fields use ``%r`` already which
escapes via repr(); no change needed there. The DEBUG-level
``code:\n%s`` line stays raw — multi-line is intentional (operator's
primary forensic artefact) and the format string already opens a
fresh line so there's nothing to inject into.

Tests
-----

* ``TestCodeModeAdditionalResourceLimits`` (3 tests) —
  memory / recursion / invocation-cap. Previous suite only covered
  timeout.
* ``TestCodeModeNormalizeEndpointTraversal`` (1 parametrized test, 4
  rows) — all 4 ``..`` traversal patterns reject.
* ``TestCodeModeProxyLaunderingBlocked`` (1 parametrized test) —
  ``call_tool`` to each of the 4 search synthetics returns the
  ``AUTH_INSUFFICIENT_PERMISSIONS`` block.
* ``test_list_saved_tools`` updated to verify new
  ``data.saved_tools[name]`` + ``data.count`` shape.
* ``test_save_warning_rollback_shape`` (skipped) — placeholder noting
  that the unit suite covers the persistence-failure rollback;
  E2E coverage requires runtime filesystem poisoning the addon
  container model doesn't expose.

Lint/mypy/ast-grep all clean. 24 unit tests still pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: kingpanther13

src/ha_mcp/server.py

@@ -547,12 +547,19 @@ def _apply_tool_search(self) -> None:
                     max_results=self.settings.tool_search_max_results,
                     always_visible=pinned,
                     search_tool_description=description,
+                    # Pinned tools must be excluded from the proxy's
+                    # category sets when code mode is on; otherwise sandbox
+                    # code can launder a recursive ``ha_manage_custom_tool``
+                    # invocation through ``ha_call_write_tool``. See the
+                    # docstring on ``_rebuild_category_cache``.
+                    enable_code_mode=self.settings.enable_code_mode,
                 )
             )
             logger.info(
-                "Tool search transform applied (%d pinned tools, max_results=%d)",
+                "Tool search transform applied (%d pinned tools, max_results=%d, code_mode=%s)",
                 len(pinned),
                 self.settings.tool_search_max_results,
+                self.settings.enable_code_mode,
             )
         except Exception:
             logger.exception("Failed to apply tool search transform")

src/ha_mcp/tools/tools_code.py

@@ -46,8 +46,45 @@
 # from disk on registration and persisted on every save_as / delete.
 _saved_tools: dict[str, dict[str, str]] = {}
 
-# Tools that sandbox code must not call (prevents recursive self-invocation)
-_BLOCKED_TOOLS = frozenset({"ha_manage_custom_tool"})
+# Translation table for ``_log_safe`` — replaces CR / LF / TAB with a single
+# space so an LLM-controlled string interpolated into a log line via ``%s``
+# can't manufacture extra log records.
+_LOG_CONTROL_CHARS_MAP = str.maketrans({"\r": " ", "\n": " ", "\t": " "})
+
+
+def _log_safe(value: Any, max_len: int = 200) -> str:
+    """Return ``value`` flattened to a single line, safe for ``%s`` log interpolation.
+
+    Replaces ``\\r`` / ``\\n`` / ``\\t`` with spaces and truncates to
+    ``max_len`` characters. Used on user-controlled strings (``justification``,
+    saved-tool ``name``, etc.) before they reach ``logger.info(..., %s, ...)``
+    so a crafted input like ``"real reason\\nFAKE_CRITICAL: thing crashed"``
+    cannot inject a second log line.
+    """
+    text = str(value)
+    if len(text) > max_len:
+        text = text[:max_len]
+    return text.translate(_LOG_CONTROL_CHARS_MAP)
+
+
+# Tools that sandbox code must not call. Includes ``ha_manage_custom_tool``
+# itself (prevents recursive self-invocation) plus the four synthetics that
+# the categorized-search transform exposes when ``ENABLE_TOOL_SEARCH=true``
+# (``ha_search_tools``, ``ha_call_{read,write,delete}_tool``). Without those
+# four entries, sandbox code could "launder" a recursive call as
+# ``call_tool("ha_call_write_tool", {"name": "ha_manage_custom_tool", ...})``
+# — the proxy would then dispatch the underlying tool and the in-sandbox
+# guard never fires. The architectural fix lives in
+# ``CategorizedSearchTransform`` (excludes pinned tools from category sets
+# when code mode is on); this set is the defense-in-depth that closes the
+# inner-call path even if the proxy is reachable some other way.
+_BLOCKED_TOOLS = frozenset({
+    "ha_manage_custom_tool",
+    "ha_search_tools",
+    "ha_call_read_tool",
+    "ha_call_write_tool",
+    "ha_call_delete_tool",
+})
 
 # Validation for save_as names
 _SAVE_NAME_PATTERN = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]{0,63}$")
@@ -99,6 +136,7 @@
     "service_executed",
     "automation_reloaded",
     "script_started",
+    "script_finished",
     "homeassistant_start",
     "homeassistant_started",
     "homeassistant_stop",
@@ -111,6 +149,11 @@
     "category_registry_updated",
     "floor_registry_updated",
     "label_registry_updated",
+    # ``logbook_entry`` is the documented logbook write API — the Logbook
+    # integration consumes it to render rows. Sandbox code firing this
+    # event would inject attacker-fabricated rows directly into the
+    # user's primary investigation tool, which is a data-integrity issue.
+    "logbook_entry",
     "lovelace_updated",
     "panels_updated",
     "themes_updated",
@@ -139,6 +182,20 @@
     "config/entity_registry/delete",
     "config/entity_registry/disable",
     "config/entity_registry/update",
+    # Floor / label / category registries follow the same rationale as
+    # area / device / entity above: each has a wrapping MCP tool
+    # (``ha_config_set_floor``, ``ha_config_set_label``,
+    # ``ha_config_set_category``) that performs invariant checks the
+    # raw WS command skips.
+    "config/floor_registry/create",
+    "config/floor_registry/delete",
+    "config/floor_registry/update",
+    "config/label_registry/create",
+    "config/label_registry/delete",
+    "config/label_registry/update",
+    "config/category_registry/create",
+    "config/category_registry/delete",
+    "config/category_registry/update",
 })
 
 
@@ -578,13 +635,19 @@ def _normalize_endpoint(endpoint: Any) -> str:
         path works whether the caller wrote ``"events"``, ``"/events"``, or
         ``"/api/events"``.
 
-        Rejects anything that looks like an absolute URL or a userinfo
-        injection: an ``://`` substring, a leading ``//`` (protocol-relative),
-        or an ``@`` before the first ``/``. Without this the httpx client
-        will dispatch the request to the absolute host *with the HA bearer
-        token still attached*, leaking credentials to whoever the LLM was
-        prompted to point at. The sandbox is supposed to be on-instance
-        only.
+        Rejects:
+
+        * Absolute URL forms — ``://``, leading ``//`` (protocol-relative),
+          or ``@`` before the first ``/`` (userinfo). Without this httpx
+          will dispatch the request to the absolute host *with the HA
+          bearer token still attached*, leaking credentials.
+        * ``..`` path segments — httpx happily resolves
+          ``base_url='http://ha:8123/api'`` + endpoint ``'../auth/providers'``
+          to ``http://ha:8123/auth/providers``, escaping the ``/api/``
+          prefix entirely. HA exposes other bearer-authenticated routes
+          at root (``/auth/...``, ``/profile``, etc.) — every one of
+          those becomes reachable from the sandbox unless we reject
+          ``..`` here.
         """
         if not isinstance(endpoint, str):
             raise ValueError("endpoint must be a string path (e.g. '/states')")
@@ -601,6 +664,16 @@ def _normalize_endpoint(endpoint: Any) -> str:
         ep = endpoint.lstrip("/")
         if ep.startswith("api/"):
             ep = ep[4:]
+        # ``..`` segments would let the sandbox escape the ``/api/`` prefix
+        # via httpx URL resolution. Check after stripping so the comparison
+        # is against actual path segments, and return the same error for
+        # leading-, mid-, or trailing-position cases.
+        for segment in ep.split("/"):
+            if segment == "..":
+                raise ValueError(
+                    "endpoint must not contain '..' path segments; "
+                    "the sandbox is restricted to /api/ routes"
+                )
         return ep
 
     async def _api_get(endpoint: str) -> Any:
@@ -992,16 +1065,26 @@ async def ha_manage_custom_tool(
 
         # --- Mode: list saved tools ---
         if list_saved:
+            # The saved-tools dict is nested under a stable ``saved_tools``
+            # key rather than spread directly under ``data`` because the
+            # name pattern (``^[a-zA-Z_][a-zA-Z0-9_]{0,63}$``) accepts
+            # values like ``result``, ``count``, ``code`` — every one of
+            # which is also a key the *other* response shapes use. A
+            # consumer reading ``r["data"]["result"]`` after a list_saved
+            # call would otherwise get a saved-tool entry instead of a
+            # run-result.
             return {
                 "success": True,
                 "data": {
-                    name: {
-                        "code": info["code"],
-                        "justification": info["justification"],
-                    }
-                    for name, info in _saved_tools.items()
+                    "saved_tools": {
+                        name: {
+                            "code": info["code"],
+                            "justification": info["justification"],
+                        }
+                        for name, info in _saved_tools.items()
+                    },
+                    "count": len(_saved_tools),
                 },
-                "count": len(_saved_tools),
             }
 
         # --- Mode: run saved tool ---
@@ -1086,7 +1169,15 @@ async def ha_manage_custom_tool(
                 )
             )
 
-        logger.info("ha_manage_custom_tool invoked — justification: %s", justification[:200])
+        logger.info(
+            "ha_manage_custom_tool invoked — justification: %s",
+            _log_safe(justification),
+        )
+        # Code is logged at DEBUG and the multi-line shape is intentional
+        # (the LLM-authored snippet is the operator's primary forensic
+        # artefact). No control-char sanitisation here because the log
+        # format string already opens a fresh line — there's nothing to
+        # inject into.
         logger.debug("ha_manage_custom_tool code:\n%s", code)
 
         try:

src/ha_mcp/transforms/categorized_search.py

@@ -166,6 +166,7 @@ def __init__(
         call_read_name: str = "ha_call_read_tool",
         call_write_name: str = "ha_call_write_tool",
         call_delete_name: str = "ha_call_delete_tool",
+        enable_code_mode: bool = False,
         **kwargs: Any,
     ) -> None:
         super().__init__(
@@ -183,6 +184,15 @@ def __init__(
         self._call_delete_name = call_delete_name
         self._search_tool_description = search_tool_description
         self._proxy_descs = _build_proxy_descriptions(search_tool_name)
+        # When code mode is enabled, the proxy must NOT dispatch to pinned
+        # tools (specifically ``ha_manage_custom_tool``) — otherwise a
+        # sandbox call to ``ha_call_write_tool`` with name=
+        # "ha_manage_custom_tool" would launder a recursive invocation
+        # past ``_BLOCKED_TOOLS`` inside the sandbox. Default False
+        # preserves existing behaviour for installations that aren't
+        # running code mode; server.py flips this on when
+        # ``settings.enable_code_mode`` is True.
+        self._enable_code_mode = enable_code_mode
 
         # Category caches rebuilt when the catalog hash changes,
         # matching BM25SearchTransform's staleness detection pattern.
@@ -201,8 +211,21 @@ def _catalog_hash(tools: Sequence[Tool]) -> str:
         return hashlib.sha256(key.encode()).hexdigest()
 
     async def _rebuild_category_cache(self, ctx: Any) -> None:
-        """Rebuild the read/write/delete category sets if catalog changed."""
-        catalog = await self.get_tool_catalog(ctx)
+        """Rebuild the read/write/delete category sets if catalog changed.
+
+        When ``self._enable_code_mode`` is True, pinned tools are excluded
+        from the category sets via ``_get_visible_tools`` (the same
+        FastMCP helper that ``BM25SearchTransform`` uses). This prevents
+        a sandbox-side recursive invocation laundered as
+        ``ha_call_write_tool(name="ha_manage_custom_tool", ...)`` —
+        without the filter, the pinned-and-callable
+        ``ha_manage_custom_tool`` ends up in ``_write_tools`` and the
+        proxy will happily dispatch.
+        """
+        if self._enable_code_mode:
+            catalog = await self._get_visible_tools(ctx)
+        else:
+            catalog = await self.get_tool_catalog(ctx)
         current_hash = self._catalog_hash(catalog)
         if current_hash == self._last_catalog_hash:
             return