From 0d4a3fa4e2621ba66a06c2afbcfb74819e69ddae Mon Sep 17 00:00:00 2001
From: Jacob Magar <jmagar@gmail.com>
Date: Sun, 15 Mar 2026 19:08:43 -0400
Subject: [PATCH] fix(live): validate log_tail path against allowlist, move
 guards before error handler

Add _ALLOWED_LOG_PREFIXES allowlist check to log_tail (mirrors storage.py pattern)
to prevent path traversal attacks. Move path/required guards before tool_error_handler
context so validation errors raise cleanly. Add two tests: ToolError propagation and
invalid path rejection.
---
 tests/test_live.py       | 19 +++++++++++++++++++
 unraid_mcp/tools/live.py | 19 +++++++++++++++++--
 2 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/tests/test_live.py b/tests/test_live.py
index 0624bcb..e303c2f 100644
--- a/tests/test_live.py
+++ b/tests/test_live.py
@@ -93,3 +93,22 @@ async def test_invalid_action_raises(mcp):
     tool_fn = _make_live_tool(mcp)
     with pytest.raises(ToolError, match="Invalid action"):
         await tool_fn(action="nonexistent")  # type: ignore[arg-type]
+
+
+@pytest.mark.asyncio
+async def test_snapshot_propagates_tool_error(mcp, _mock_subscribe_once):
+    from unraid_mcp.core.exceptions import ToolError
+
+    _mock_subscribe_once.side_effect = ToolError("Subscription timed out after 10s")
+    tool_fn = _make_live_tool(mcp)
+    with pytest.raises(ToolError, match="timed out"):
+        await tool_fn(action="cpu")
+
+
+@pytest.mark.asyncio
+async def test_log_tail_rejects_invalid_path(mcp, _mock_subscribe_collect):
+    from unraid_mcp.core.exceptions import ToolError
+
+    tool_fn = _make_live_tool(mcp)
+    with pytest.raises(ToolError, match="must start with"):
+        await tool_fn(action="log_tail", path="/etc/shadow")
diff --git a/unraid_mcp/tools/live.py b/unraid_mcp/tools/live.py
index caed254..b02ac34 100644
--- a/unraid_mcp/tools/live.py
+++ b/unraid_mcp/tools/live.py
@@ -10,6 +10,7 @@ Use `subscribe_collect` actions for event streams (notification_feed, log_tail).
 
 from __future__ import annotations
 
+import os
 from typing import Any, Literal, get_args
 
 from fastmcp import FastMCP
@@ -19,6 +20,8 @@ from ..core.exceptions import ToolError, tool_error_handler
 from ..subscriptions.snapshot import subscribe_collect, subscribe_once
 
 
+_ALLOWED_LOG_PREFIXES = ("/var/log/", "/boot/logs/", "/mnt/")
+
 SNAPSHOT_ACTIONS = {
     "cpu": """
         subscription { systemMetricsCpu { id percentTotal cpus { percentTotal percentUser percentSystem percentIdle } } }
@@ -122,6 +125,20 @@ def register_live_tool(mcp: FastMCP) -> None:
                 f"Invalid action '{action}'. Must be one of: {sorted(ALL_LIVE_ACTIONS)}"
             )
 
+        # Validate log_tail path before entering the error handler context.
+        if action == "log_tail":
+            if not path:
+                raise ToolError("path is required for 'log_tail' action")
+            # Resolve to prevent path traversal attacks (same as storage.py).
+            # Using os.path.realpath instead of anyio.Path.resolve() because the
+            # async variant blocks on NFS-mounted paths under /mnt/ (Perf-AI-1).
+            normalized = os.path.realpath(path)  # noqa: ASYNC240
+            if not any(normalized.startswith(p) for p in _ALLOWED_LOG_PREFIXES):
+                raise ToolError(
+                    f"path must start with one of: {', '.join(_ALLOWED_LOG_PREFIXES)}. Got: {path!r}"
+                )
+            path = normalized
+
         with tool_error_handler("live", action, logger):
             logger.info(f"Executing unraid_live action={action} timeout={timeout}")
 
@@ -131,8 +148,6 @@ def register_live_tool(mcp: FastMCP) -> None:
 
             # Collect actions
             if action == "log_tail":
-                if not path:
-                    raise ToolError("path is required for 'log_tail' action")
                 events = await subscribe_collect(
                     COLLECT_ACTIONS["log_tail"],
                     variables={"path": path},