Merge branch 'main' into bump-kv

2026-08-21 04:54:17 +02:00 · 2025-10-14 16:52:24 -05:00 · 2025-10-14 16:52:24 -05:00 · ea9b2de8f8
commit ea9b2de8f8
parent 5d13deffc9 5f8461b7f0
8 changed files with 177 additions and 15 deletions
--- a/src/fastmcp/cli/cli.py
+++ b/src/fastmcp/cli/cli.py
@ -46,9 +46,7 @@ def _get_npx_command():
        # Try both npx.cmd and npx.exe on Windows
        for cmd in ["npx.cmd", "npx.exe", "npx"]:
            try:
-                subprocess.run(
-                    [cmd, "--version"], check=True, capture_output=True, shell=True
-                )
+                subprocess.run([cmd, "--version"], check=True, capture_output=True)
                return cmd
            except subprocess.CalledProcessError:
                continue
@ -277,12 +275,10 @@ async def dev(
        # Set marker to prevent infinite loops when subprocess calls FastMCP
        env = dict(os.environ.items()) | env_vars | {"FASTMCP_UV_SPAWNED": "1"}

-        # Run the MCP Inspector command with shell=True on Windows
-        shell = sys.platform == "win32"
+        # Run the MCP Inspector command
        process = subprocess.run(
            [npx_cmd, inspector_cmd] + uv_cmd,
            check=True,
-            shell=shell,
            env=env,
        )
        sys.exit(process.returncode)
--- a/src/fastmcp/cli/install/cursor.py
+++ b/src/fastmcp/cli/install/cursor.py
@ -56,7 +56,7 @@ def open_deeplink(deeplink: str) -> bool:
            subprocess.run(["open", deeplink], check=True, capture_output=True)
        elif sys.platform == "win32":  # Windows
            subprocess.run(
-                ["start", deeplink], shell=True, check=True, capture_output=True
+                ["cmd", "/c", "start", deeplink], check=True, capture_output=True
            )
        else:  # Linux and others
            subprocess.run(["xdg-open", deeplink], check=True, capture_output=True)
--- a/src/fastmcp/client/oauth_callback.py
+++ b/src/fastmcp/client/oauth_callback.py
@ -46,9 +46,7 @@ def create_callback_html(
    # Add detail info box for both success and error cases
    detail_info = ""
    if is_success and server_url:
-        detail_info = create_info_box(
-            f"Connected to: <strong>{server_url}</strong>", centered=True
-        )
+        detail_info = create_info_box(f"Connected to: {server_url}", centered=True)
    elif not is_success:
        detail_info = create_info_box(message, is_error=True, centered=True)

--- a/src/fastmcp/server/auth/providers/jwt.py
+++ b/src/fastmcp/server/auth/providers/jwt.py
@ -382,7 +382,12 @@ class JWTVerifier(TokenVerifier):
            claims = self.jwt.decode(token, verification_key)

            # Extract client ID early for logging
-            client_id = claims.get("client_id") or claims.get("sub") or "unknown"
+            client_id = (
+                claims.get("client_id")
+                or claims.get("azp")
+                or claims.get("sub")
+                or "unknown"
+            )

            # Validate expiration
            exp = claims.get("exp")
--- a/src/fastmcp/utilities/ui.py
+++ b/src/fastmcp/utilities/ui.py
@ -7,6 +7,8 @@ consent pages, and other user-facing interfaces.

 from __future__ import annotations

+import html
+
 from starlette.responses import HTMLResponse

 # FastMCP branding
@ -339,6 +341,7 @@ def create_page(
    Returns:
        Complete HTML page as string
    """
+    title = html.escape(title)
    return f"""
    <!DOCTYPE html>
    <html lang="en">
@ -375,6 +378,7 @@ def create_status_message(message: str, is_success: bool = True) -> str:
    Returns:
        HTML for status message
    """
+    message = html.escape(message)
    icon = "✓" if is_success else "✕"
    icon_class = "success" if is_success else "error"

@ -400,6 +404,7 @@ def create_info_box(
    Returns:
        HTML for info box
    """
+    content = html.escape(content)
    classes = ["info-box"]
    if is_error:
        classes.append("error")
@ -422,8 +427,8 @@ def create_detail_box(rows: list[tuple[str, str]]) -> str:
    rows_html = "\n".join(
        f"""
        <div class="detail-row">
-            <div class="detail-label">{label}:</div>
-            <div class="detail-value">{value}</div>
+            <div class="detail-label">{html.escape(label)}:</div>
+            <div class="detail-value">{html.escape(value)}</div>
        </div>
        """
        for label, value in rows
--- a/tests/cli/test_cli.py
+++ b/tests/cli/test_cli.py
@ -416,7 +416,6 @@ class TestWindowsSpecific:
                ["npx.cmd", "--version"],
                check=True,
                capture_output=True,
-                shell=True,
            )

    @patch("subprocess.run")
--- a/tests/cli/test_cursor.py
+++ b/tests/cli/test_cursor.py
@ -145,7 +145,7 @@ class TestOpenDeeplink:

            assert result is True
            mock_run.assert_called_once_with(
-                ["start", "cursor://test"], shell=True, check=True, capture_output=True
+                ["cmd", "/c", "start", "cursor://test"], check=True, capture_output=True
            )

    @patch("subprocess.run")
--- a/tests/client/test_oauth_callback_xss.py
+++ b/tests/client/test_oauth_callback_xss.py
@ -0,0 +1,159 @@
+"""Comprehensive XSS protection tests for OAuth callback HTML rendering."""
+
+import pytest
+
+from fastmcp.client.oauth_callback import create_callback_html
+from fastmcp.utilities.ui import (
+    create_detail_box,
+    create_info_box,
+    create_page,
+    create_status_message,
+)
+
+
+def test_ui_create_page_escapes_title():
+    """Test that page title is properly escaped."""
+    xss_title = "<script>alert(1)</script>"
+    html = create_page("content", title=xss_title)
+    assert "&lt;script&gt;alert(1)&lt;/script&gt;" in html
+    assert "<script>alert(1)</script>" not in html
+
+
+def test_ui_create_status_message_escapes():
+    """Test that status messages are properly escaped."""
+    xss_message = "<img src=x onerror=alert(1)>"
+    html = create_status_message(xss_message)
+    assert "&lt;img src=x onerror=alert(1)&gt;" in html
+    assert "<img src=x onerror=alert(1)>" not in html
+
+
+def test_ui_create_info_box_escapes():
+    """Test that info box content is properly escaped."""
+    xss_content = "<iframe src=javascript:alert(1)></iframe>"
+    html = create_info_box(xss_content)
+    assert "&lt;iframe" in html
+    assert "<iframe src=javascript:alert(1)>" not in html
+
+
+def test_ui_create_detail_box_escapes():
+    """Test that detail box labels and values are properly escaped."""
+    xss_label = '<script>alert("label")</script>'
+    xss_value = '<script>alert("value")</script>'
+    html = create_detail_box([(xss_label, xss_value)])
+    assert "&lt;script&gt;" in html
+    assert '<script>alert("label")</script>' not in html
+    assert '<script>alert("value")</script>' not in html
+
+
+def test_callback_html_escapes_error_message():
+    """Test that XSS payloads in error messages are properly escaped."""
+    xss_payload = "<img/src/onerror=alert(1)>"
+    html = create_callback_html(xss_payload, is_success=False)
+
+    assert "&lt;img/src/onerror=alert(1)&gt;" in html
+    assert "<img/src/onerror=alert(1)>" not in html
+
+
+def test_callback_html_escapes_server_url():
+    """Test that XSS payloads in server_url are properly escaped."""
+    xss_payload = "<script>alert(1)</script>"
+    html = create_callback_html("Success", is_success=True, server_url=xss_payload)
+
+    assert "&lt;script&gt;alert(1)&lt;/script&gt;" in html
+    assert "<script>alert(1)</script>" not in html
+
+
+def test_callback_html_escapes_title():
+    """Test that XSS payloads in title are properly escaped."""
+    xss_payload = "<script>alert(document.domain)</script>"
+    html = create_callback_html("Success", title=xss_payload)
+
+    assert "&lt;script&gt;alert(document.domain)&lt;/script&gt;" in html
+    assert "<script>alert(document.domain)</script>" not in html
+
+
+def test_callback_html_mixed_content():
+    """Test that legitimate text mixed with XSS attempts is properly escaped."""
+    mixed_payload = "Error: <img src=x onerror=alert(1)> occurred"
+    html = create_callback_html(mixed_payload, is_success=False)
+
+    assert "&lt;img src=x onerror=alert(1)&gt;" in html
+    assert "Error:" in html
+    assert "occurred" in html
+    assert "<img src=x onerror=alert(1)>" not in html
+
+
+def test_callback_html_event_handlers():
+    """Test that event handler attributes are escaped."""
+    xss_payload = '" onload="alert(1)'
+    html = create_callback_html(xss_payload, is_success=False)
+
+    assert "&quot; onload=&quot;alert(1)" in html
+    assert '" onload="alert(1)' not in html
+
+
+def test_callback_html_special_characters():
+    """Test that special HTML characters are properly escaped."""
+    special_chars = "&<>\"'/"
+    html = create_callback_html(special_chars, is_success=False)
+
+    assert "&amp;" in html
+    assert "&lt;" in html
+    assert "&gt;" in html
+    assert "&quot;" in html
+    # Apostrophe gets escaped to &#x27; by html.escape()
+    assert "&#x27;" in html
+
+
+@pytest.mark.parametrize(
+    "xss_vector",
+    [
+        "<img src=x onerror=alert(1)>",
+        "<script>alert(document.cookie)</script>",
+        "<iframe src=javascript:alert(1)>",
+        "<svg/onload=alert(1)>",
+        "<body onload=alert(1)>",
+        "<input onfocus=alert(1) autofocus>",
+        "<select onfocus=alert(1) autofocus>",
+        "<textarea onfocus=alert(1) autofocus>",
+        "<marquee onstart=alert(1)>",
+        "<div style=background:url('javascript:alert(1)')>",
+    ],
+)
+def test_common_xss_vectors(xss_vector: str):
+    """Test that common XSS attack vectors are properly escaped."""
+    html = create_callback_html(xss_vector, is_success=False)
+
+    # Should not contain the raw XSS vector
+    assert xss_vector not in html
+
+    # Should contain escaped version (at least the < and > should be escaped)
+    assert "&lt;" in html
+    assert "&gt;" in html
+
+
+def test_legitimate_content_still_works():
+    """Ensure legitimate content is displayed correctly after escaping."""
+    legitimate_message = "Authentication failed: Invalid credentials"
+    legitimate_url = "https://example.com:8080/mcp"
+
+    # Error case
+    html = create_callback_html(legitimate_message, is_success=False)
+    assert legitimate_message in html
+    assert "Authentication failed" in html
+
+    # Success case
+    html = create_callback_html("Success", is_success=True, server_url=legitimate_url)
+    assert legitimate_url in html
+    assert "Authentication successful" in html
+
+
+def test_no_hardcoded_html_tags():
+    """Verify that there are no hardcoded HTML tags that bypass escaping."""
+    server_url = "test-server"
+    html = create_callback_html("Success", is_success=True, server_url=server_url)
+
+    # Should not have <strong> tags around the server URL
+    assert f"<strong>{server_url}</strong>" not in html
+    # Should have the server URL displayed normally (escaped)
+    assert server_url in html