mcp: address PR #4508 review feedback

- log_analysis: exact-match portal hostname (CodeQL); safer tar/zip
  extraction; honor SAS URL instead of DefaultAzureCredential.
- server.py: FORWARDED_ALLOW_IPS env var (default 127.0.0.1).
- Dockerfile: run as non-root mcp user; add .dockerignore.
- docs_index.yaml: remove duplicate keys.
- tests: reconcile tool count to 25; fix trivial assertions; --xml help.

Author: Johnson George

mcp/.dockerignore

@@ -0,0 +1,22 @@
+# Build / packaging artifacts
+**/__pycache__
+**/*.pyc
+**/*.pyo
+**/*.egg-info
+build/
+dist/
+
+# Virtualenvs and editor state
+.venv/
+venv/
+.env
+.env.*
+.vscode/
+.idea/
+
+# Internal / local-only scripts and scratch files (gitignored)
+tmp/
+
+# Tests are still copied (used by smoke tests in CI), but exclude fixtures
+# generated locally:
+tests/__pycache__/

mcp/Dockerfile

@@ -20,6 +20,12 @@ RUN pip install --no-cache-dir "/app/lisa/mcp[azure]"
 # Point the MCP server at the cloned repo
 ENV LISA_REPO_ROOT=/app/lisa
 
+# Drop privileges: run as a non-root user. /app must remain readable but
+# writable areas (logs, downloads) are scoped to /home/mcp at runtime.
+RUN useradd --create-home --shell /usr/sbin/nologin --uid 10001 mcp \
+    && chown -R mcp:mcp /app
+USER mcp
+
 EXPOSE 8080
 
 ENTRYPOINT ["lisa-mcp"]

mcp/lisa_mcp/docs_index.yaml

@@ -90,37 +90,6 @@ tools:
       - docs/write_test/write_case.rst
 
   # ── knowledge.py ───────────────────────────────────────────────
-  lisa_explain_concept:
-    primary: docs/write_test/concepts.rst
-    supplementary: []
-
-  lisa_get_api_reference:
-    primary: docs/write_test/write_case.rst
-    supplementary: []
-
-  lisa_find_examples:
-    primary: docs/write_test/write_case.rst
-    supplementary: []
-
-  lisa_list_tools:
-    primary: docs/write_test/write_case.rst
-    supplementary: []
-
-  lisa_list_features:
-    primary: docs/write_test/write_case.rst
-    supplementary: []
-
-  lisa_explain_error:
-    primary: docs/run_test/troubleshoot_failures.rst
-    supplementary: []
-
-  # ── execution.py ───────────────────────────────────────────────
-  lisa_run:
-    primary: docs/run_test/run.rst
-    supplementary:
-      - docs/run_test/command_line.rst
-
-  # ── Framework Knowledge ────────────────────────────────────────
   lisa_explain_concept:
     primary: docs/write_test/concepts.rst
     supplementary:
@@ -146,6 +115,16 @@ tools:
     primary: docs/write_test/extension.rst
     supplementary: []
 
+  lisa_explain_error:
+    primary: docs/run_test/troubleshoot_failures.rst
+    supplementary: []
+
+  # ── execution.py ───────────────────────────────────────────────
+  lisa_run:
+    primary: docs/run_test/run.rst
+    supplementary:
+      - docs/run_test/command_line.rst
+
 
 # ── Topic index (used by explain_concept for targeted lookup) ────
 #

mcp/lisa_mcp/server.py

@@ -126,12 +126,18 @@ async def handle_sse(request):
             ],
         )
 
+        # Restrict which proxy IPs may set X-Forwarded-* headers. Set
+        # FORWARDED_ALLOW_IPS to your reverse proxy's IP(s) in deployment.
+        # Defaults to loopback to prevent client-side spoofing of the
+        # forwarded client IP.
+        forwarded_allow_ips = os.environ.get("FORWARDED_ALLOW_IPS", "127.0.0.1")
+
         uvicorn.run(
             app,
             host=args.host,
             port=args.port,
             log_level="info",
-            forwarded_allow_ips="*",
+            forwarded_allow_ips=forwarded_allow_ips,
             proxy_headers=True,
         )
     else:

mcp/lisa_mcp/tools/log_analysis.py

@@ -425,9 +425,13 @@ def lisa_download_logs(
         is_azure_blob = parsed.hostname and parsed.hostname.endswith(
             ".blob.core.windows.net"
         )
+        # If the URL already carries a SAS token, the URL itself is the
+        # credential — skip the Azure SDK / DefaultAzureCredential path
+        # and let the plain HTTPS download below use the SAS URL directly.
+        has_sas = "sig=" in (parsed.query or "")
 
         # Azure blob prefix (virtual directory) — list + download all
-        if is_azure_blob and not auth_token:
+        if is_azure_blob and not auth_token and not has_sas:
             path_parts = [p for p in parsed.path.strip("/").split("/") if p]
             if len(path_parts) >= 2:
                 container = path_parts[0]
@@ -1272,7 +1276,8 @@ def _parse_portal_storage_url(url: str) -> Optional[dict[str, str]]:
     Returns ``None`` if the URL is not a portal storage URL.
     """
     parsed = urlparse(url)
-    if not parsed.hostname or not parsed.hostname.endswith("portal.azure.com"):
+    host = (parsed.hostname or "").rstrip(".").lower()
+    if host != "portal.azure.com":
         return None
     if not parsed.fragment:
         return None
@@ -1397,25 +1402,32 @@ def _extract_archive(download_path: str, download_dir: str) -> str:
 
     if tarfile.is_tarfile(download_path):
         os.makedirs(extract_dir, exist_ok=True)
+        abs_extract = os.path.abspath(extract_dir)
         with tarfile.open(download_path) as tf:
-            safe_members = [
-                m
-                for m in tf.getmembers()
-                if not m.name.startswith(("/", "..")) and ".." not in m.name
-            ]
-            tf.extractall(extract_dir, members=safe_members)
+            safe_members = []
+            for m in tf.getmembers():
+                target = os.path.abspath(os.path.join(abs_extract, m.name))
+                if os.path.commonpath([abs_extract, target]) != abs_extract:
+                    continue
+                safe_members.append(m)
+            # filter="data" (PEP 706) blocks unsafe members (links, abs paths,
+            # device files) on Python 3.12+; older versions ignore the kwarg
+            # via the try/except.
+            try:
+                tf.extractall(extract_dir, members=safe_members, filter="data")
+            except TypeError:
+                tf.extractall(extract_dir, members=safe_members)
         os.remove(download_path)
         return extract_dir
 
     if zipfile.is_zipfile(download_path):
         os.makedirs(extract_dir, exist_ok=True)
+        abs_extract = os.path.abspath(extract_dir)
         with zipfile.ZipFile(download_path) as zf:
-            safe_names = [
-                n
-                for n in zf.namelist()
-                if not n.startswith(("/", "..")) and ".." not in n
-            ]
-            for name in safe_names:
+            for name in zf.namelist():
+                target = os.path.abspath(os.path.join(abs_extract, name))
+                if os.path.commonpath([abs_extract, target]) != abs_extract:
+                    continue
                 zf.extract(name, extract_dir)
         os.remove(download_path)
         return extract_dir

mcp/run_tests.py

@@ -70,12 +70,12 @@ def main() -> int:
     group.add_argument(
         "--smoke",
         action="store_true",
-        help="Quick smoke test — verify all 24 tools are registered",
+        help="Quick smoke test — verify all 25 tools are registered",
     )
     parser.add_argument(
         "--xml",
         action="store_true",
-        help="Output JUnit XML report to test-results.xml",
+        help="Output JUnit XML reports into the test-results/ directory",
     )
     parser.add_argument(
         "-v",

mcp/tests/test_all_tools.py

@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
-"""Comprehensive functional tests for all 24 MCP tools.
+"""Comprehensive functional tests for all 25 MCP tools.
 
 Run from the mcp/ directory:
     python -m pytest tests/test_all_tools.py -v
@@ -295,8 +295,8 @@ def test_log_from_content(self) -> None:
             "lisa_analyze_log",
             log_content="smoke_test | PASSED | ok\nverify_x | FAILED | boom\n",
         )
-        self.assertIn("1", result)  # 1 passed
-        self.assertIn("1", result)  # 1 failed
+        self.assertIn("passed", result.lower())
+        self.assertIn("1 failed", result.lower())
 
     def test_empty_log(self) -> None:
         result = _call("lisa_analyze_log", log_content="nothing relevant here\n")
@@ -726,7 +726,7 @@ def test_returns_features(self) -> None:
 
 
 # ======================================================================
-# Cross-cutting: verify all 24 tools are registered
+# Cross-cutting: verify all 25 tools are registered
 # ======================================================================
 
 

mcp/tests/test_mcp_integration.py

@@ -80,7 +80,7 @@ async def _list_tools(self) -> list:
     def test_server_starts_and_lists_tools(self) -> None:
         tools = _run(self._list_tools())
         names = {t.name for t in tools}
-        self.assertEqual(len(names), 24, f"Expected 24 tools, got {len(names)}")
+        self.assertEqual(len(names), 25, f"Expected 25 tools, got {len(names)}")
         self.assertIn("lisa_analyze_log", names)
         self.assertIn("lisa_write_test", names)
         self.assertIn("lisa_explain_concept", names)