Merge pull request #30 from signnow/fix/cors-expose-headers-preflight

Expose Mcp-Session-Id header in CORS preflight responses

Author: mihasicehcek

src/signnow_client/models/folders_lite.py

@@ -44,6 +44,11 @@ def _normalize_folder_type_value(value: Any) -> Any:
     return value
 
 
+def _normalize_to_unknown(_: object) -> str:
+    """Always return 'unknown' — used as BeforeValidator for UnknownFolderDocLite.type."""
+    return "unknown"
+
+
 def _normalize_roles(value: Any) -> list[str] | None:
     """Normalize roles to list[str] format.
 
@@ -171,7 +176,6 @@ class DocumentGroupDocumentLite(SNBaseModel):
 DocTypeTemplate = Annotated[Literal["template"], BeforeValidator(_normalize_folder_type_value)]
 DocTypeDocGroup = Annotated[Literal["document-group"], BeforeValidator(_normalize_folder_type_value)]
 DocTypeDgt = Annotated[Literal["dgt"], BeforeValidator(_normalize_folder_type_value)]
-DocTypeUnknown = Annotated[Literal["unknown"], BeforeValidator(_normalize_folder_type_value)]
 
 
 class DocumentItemLite(SNBaseModel):
@@ -272,7 +276,7 @@ class DocumentGroupTemplateItemLite(SNBaseModel):
 class UnknownFolderDocLite(SNBaseModel):
     """Fallback model for folder items with unknown or missing type/entity_type."""
 
-    type: DocTypeUnknown = Field(..., validation_alias=AliasChoices("type", "entity_type"))
+    type: Annotated[Literal["unknown"], BeforeValidator(_normalize_to_unknown)] = Field("unknown", validation_alias=AliasChoices("type", "entity_type"))
 
     id: str
     user_id: str | None = None

src/sn_mcp_server/app.py

@@ -1,7 +1,11 @@
+from collections.abc import Sequence
+from typing import Any
+
 from fastmcp.server.http import create_sse_app
 from starlette.applications import Starlette
 from starlette.middleware.cors import CORSMiddleware
 from starlette.routing import Mount, Route
+from starlette.types import ASGIApp
 
 from .auth import (
     BearerJWTASGIMiddleware,
@@ -11,6 +15,20 @@
 from .config import load_settings
 
 
+class _CORSMiddlewareWithExposeInPreflight(CORSMiddleware):
+    """
+    Starlette's CORSMiddleware adds Access-Control-Expose-Headers only to
+    actual (non-preflight) responses. This subclass also injects it into
+    the preflight response headers so that clients such as Claude's MCP
+    web client can read Mcp-Session-Id after the OPTIONS handshake.
+    """
+
+    def __init__(self, app: ASGIApp, expose_headers: Sequence[str] = (), **kwargs: Any) -> None:  # noqa: ANN401
+        super().__init__(app, expose_headers=expose_headers, **kwargs)
+        if expose_headers:
+            self.preflight_headers["Access-Control-Expose-Headers"] = ", ".join(expose_headers)
+
+
 def create_http_app() -> Starlette:
     """Create and configure Starlette HTTP application with MCP endpoints"""
     # ============= CONFIG =============
@@ -41,11 +59,11 @@ def create_http_app() -> Starlette:
 
     # CORS for browser clients (inspector) - BEFORE BearerJWTMiddleware
     app.add_middleware(
-        CORSMiddleware,
+        _CORSMiddlewareWithExposeInPreflight,
         allow_origins=["*"],
         allow_methods=["*"],
         allow_headers=["*"],
-        expose_headers=["*"],
+        expose_headers=["Mcp-Session-Id"],
         allow_credentials=True,
     )
 

src/sn_mcp_server/tools/list_templates.py

@@ -78,7 +78,7 @@ async def _list_all_templates(ctx: Context, token: str, client: SignNowAPIClient
                             roles=role_names,
                         )
                     )
-    except (ValueError, KeyError, AttributeError):
+    except Exception:  # noqa: S110
         # Skip root folder if it can't be accessed
         pass
 
@@ -120,9 +120,8 @@ async def _list_all_templates(ctx: Context, token: str, client: SignNowAPIClient
                                 roles=role_names,
                             )
                         )
-        except (ValueError, KeyError, AttributeError):
-            # Skip folders that can't be accessed
-            # Log specific error types but continue processing other folders
+        except Exception:  # noqa: S112
+            # Skip folders that can't be accessed; continue processing remaining folders
             continue
 
     # Get template groups

tests/unit/signnow_client/test_invite_from_alias.py

@@ -12,10 +12,17 @@
 
 
 class _DummyDocumentClient(DocumentClientMixin):
-    def __init__(self):
+    def __init__(self) -> None:
         self.last_post = None
 
-    def _post(self, url: str, headers=None, data=None, json_data=None, validate_model=None):
+    def _post(  # noqa: ANN202
+        self,
+        url: str,
+        headers: dict | None = None,
+        data: dict | None = None,
+        json_data: dict | None = None,
+        validate_model: type | None = None,
+    ) -> object:
         self.last_post = {"url": url, "headers": headers, "data": data, "json_data": json_data, "validate_model": validate_model}
         if validate_model is None:
             return None
@@ -27,7 +34,7 @@ def _post(self, url: str, headers=None, data=None, json_data=None, validate_mode
         return validate_model.model_validate({})
 
 
-def test_create_document_field_invite_request_accepts_from_field_name_and_serializes_alias():
+def test_create_document_field_invite_request_accepts_from_field_name_and_serializes_alias() -> None:
     req = CreateDocumentFieldInviteRequest(document_id="doc123", to=[], from_="sample-apps@signnow.com")
     assert req.from_ == "sample-apps@signnow.com"
 
@@ -36,7 +43,7 @@ def test_create_document_field_invite_request_accepts_from_field_name_and_serial
     assert "from_" not in dumped
 
 
-def test_create_document_freeform_invite_request_accepts_from_field_name_and_serializes_alias():
+def test_create_document_freeform_invite_request_accepts_from_field_name_and_serializes_alias() -> None:
     req = CreateDocumentFreeformInviteRequest(to="signer@example.com", from_="sample-apps@signnow.com")
     assert req.from_ == "sample-apps@signnow.com"
 
@@ -45,19 +52,19 @@ def test_create_document_freeform_invite_request_accepts_from_field_name_and_ser
     assert "from_" not in dumped
 
 
-def test_client_document_field_invite_uses_by_alias_true_when_dumping():
+def test_client_document_field_invite_uses_by_alias_true_when_dumping() -> None:
     client = _DummyDocumentClient()
     request_data = Mock()
     request_data.model_dump.return_value = {"from": "sample-apps@signnow.com"}
 
-    client.create_document_field_invite(token="t", document_id="doc123", request_data=request_data)
+    client.create_document_field_invite(token="t", document_id="doc123", request_data=request_data)  # noqa: S106
     request_data.model_dump.assert_called_once_with(exclude_none=True, by_alias=True)
 
 
-def test_client_document_freeform_invite_uses_by_alias_true_when_dumping():
+def test_client_document_freeform_invite_uses_by_alias_true_when_dumping() -> None:
     client = _DummyDocumentClient()
     request_data = Mock()
     request_data.model_dump.return_value = {"from": "sample-apps@signnow.com"}
 
-    client.create_document_freeform_invite(token="t", document_id="doc123", request_data=request_data)
+    client.create_document_freeform_invite(token="t", document_id="doc123", request_data=request_data)  # noqa: S106
     request_data.model_dump.assert_called_once_with(exclude_none=True, by_alias=True)

tests/unit/sn_mcp_server/test_cors_expose_headers.py

@@ -0,0 +1,82 @@
+"""Unit tests for _CORSMiddlewareWithExposeInPreflight in app module."""
+
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import PlainTextResponse
+from starlette.testclient import TestClient
+
+from sn_mcp_server.app import _CORSMiddlewareWithExposeInPreflight
+
+_ORIGIN = "https://claude.ai"
+_PREFLIGHT_HEADERS = {
+    "Origin": _ORIGIN,
+    "Access-Control-Request-Method": "POST",
+    "Access-Control-Request-Headers": "content-type,authorization,mcp-session-id",
+}
+
+
+def _make_app(expose_headers: list[str]) -> TestClient:
+    async def _handler(request: Request) -> PlainTextResponse:
+        return PlainTextResponse("OK")
+
+    app = Starlette()
+    app.add_middleware(
+        _CORSMiddlewareWithExposeInPreflight,
+        allow_origins=["*"],
+        allow_methods=["*"],
+        allow_headers=["*"],
+        expose_headers=expose_headers,
+        allow_credentials=True,
+    )
+    app.add_route("/mcp", _handler, methods=["OPTIONS", "POST", "GET"])
+    return TestClient(app, raise_server_exceptions=True)
+
+
+class TestCORSExposeHeadersPreflight:
+    """Regression tests for Access-Control-Expose-Headers in OPTIONS preflight responses."""
+
+    def test_preflight_includes_expose_header(self) -> None:
+        """OPTIONS response must carry Access-Control-Expose-Headers: Mcp-Session-Id."""
+        client = _make_app(["Mcp-Session-Id"])
+        resp = client.options("/mcp", headers=_PREFLIGHT_HEADERS)
+        assert resp.status_code == 200
+        assert resp.headers.get("access-control-expose-headers") == "Mcp-Session-Id"
+
+    def test_actual_response_includes_expose_header(self) -> None:
+        """Non-preflight POST response must also carry Access-Control-Expose-Headers."""
+        client = _make_app(["Mcp-Session-Id"])
+        resp = client.post("/mcp", headers={"Origin": _ORIGIN})
+        assert resp.headers.get("access-control-expose-headers") == "Mcp-Session-Id"
+
+    def test_expose_header_absent_when_empty_list(self) -> None:
+        """When expose_headers=[], Access-Control-Expose-Headers must not be present."""
+        client = _make_app([])
+        resp = client.options("/mcp", headers=_PREFLIGHT_HEADERS)
+        assert "access-control-expose-headers" not in resp.headers
+
+    def test_preflight_echoes_request_origin_with_credentials(self) -> None:
+        """With allow_credentials=True the preflight must echo the exact request origin."""
+        client = _make_app(["Mcp-Session-Id"])
+        resp = client.options("/mcp", headers=_PREFLIGHT_HEADERS)
+        assert resp.headers.get("access-control-allow-origin") == _ORIGIN
+        assert resp.headers.get("access-control-allow-credentials") == "true"
+
+    def test_preflight_multiple_expose_headers(self) -> None:
+        """Multiple expose_headers values are joined with ', ' in the preflight."""
+        client = _make_app(["Mcp-Session-Id", "X-Custom-Header"])
+        resp = client.options("/mcp", headers=_PREFLIGHT_HEADERS)
+        expose = resp.headers.get("access-control-expose-headers", "")
+        assert "Mcp-Session-Id" in expose
+        assert "X-Custom-Header" in expose
+
+    def test_preflight_returns_200(self) -> None:
+        """Preflight response status must be 200 OK."""
+        client = _make_app(["Mcp-Session-Id"])
+        resp = client.options("/mcp", headers=_PREFLIGHT_HEADERS)
+        assert resp.status_code == 200
+
+    def test_max_age_present_in_preflight(self) -> None:
+        """access-control-max-age must be set in preflight response."""
+        client = _make_app(["Mcp-Session-Id"])
+        resp = client.options("/mcp", headers=_PREFLIGHT_HEADERS)
+        assert resp.headers.get("access-control-max-age") == "600"

tests/unit/sn_mcp_server/tools/test_expiration.py

@@ -5,14 +5,13 @@
 import pytest
 
 from signnow_client.models.document_groups import DocumentGroupV2FieldInvite
-
 from sn_mcp_server.tools.models import SimplifiedInviteParticipant
 
 
 class TestExpirationHandling:
     """Test cases for optional expiration_time and expiration_days."""
 
-    def test_simplified_invite_participant_from_field_invite_with_expiration(self):
+    def test_simplified_invite_participant_from_field_invite_with_expiration(self) -> None:
         """Test SimplifiedInviteParticipant.from_document_group_v2_field_invite with expiration."""
         now = 1500000000
         field_invite = DocumentGroupV2FieldInvite(
@@ -23,7 +22,7 @@ def test_simplified_invite_participant_from_field_invite_with_expiration(self):
             expiration_time=2000000000,  # Future expiration
             expiration_days=30,
             signer_email="test@example.com",
-            password_protected="0",
+            password_protected="0",  # noqa: S106
             email_statuses=[],
         )
         participant = SimplifiedInviteParticipant.from_document_group_v2_field_invite(field_invite, now)
@@ -32,7 +31,7 @@ def test_simplified_invite_participant_from_field_invite_with_expiration(self):
         assert participant.expired is False  # Not expired yet
         assert participant.status == "pending"
 
-    def test_simplified_invite_participant_from_field_invite_without_expiration(self):
+    def test_simplified_invite_participant_from_field_invite_without_expiration(self) -> None:
         """Test SimplifiedInviteParticipant.from_document_group_v2_field_invite without expiration."""
         now = 1500000000
         field_invite = DocumentGroupV2FieldInvite(
@@ -43,7 +42,7 @@ def test_simplified_invite_participant_from_field_invite_without_expiration(self
             expiration_time=None,  # No expiration
             expiration_days=None,
             signer_email="test@example.com",
-            password_protected="0",
+            password_protected="0",  # noqa: S106
             email_statuses=[],
         )
         participant = SimplifiedInviteParticipant.from_document_group_v2_field_invite(field_invite, now)
@@ -52,7 +51,7 @@ def test_simplified_invite_participant_from_field_invite_without_expiration(self
         assert participant.expired is False  # No expiration means not expired
         assert participant.status == "pending"
 
-    def test_simplified_invite_participant_from_field_invite_expired(self):
+    def test_simplified_invite_participant_from_field_invite_expired(self) -> None:
         """Test SimplifiedInviteParticipant.from_document_group_v2_field_invite with expired invite."""
         now = 2500000000  # After expiration
         field_invite = DocumentGroupV2FieldInvite(
@@ -63,7 +62,7 @@ def test_simplified_invite_participant_from_field_invite_expired(self):
             expiration_time=2000000000,  # Past expiration
             expiration_days=30,
             signer_email="test@example.com",
-            password_protected="0",
+            password_protected="0",  # noqa: S106
             email_statuses=[],
         )
         participant = SimplifiedInviteParticipant.from_document_group_v2_field_invite(field_invite, now)
@@ -72,7 +71,7 @@ def test_simplified_invite_participant_from_field_invite_expired(self):
         assert participant.expired is True  # Expired
         assert participant.status == "expired"
 
-    def test_simplified_invite_participant_from_field_invite_expired_status(self):
+    def test_simplified_invite_participant_from_field_invite_expired_status(self) -> None:
         """Test SimplifiedInviteParticipant.from_document_group_v2_field_invite with expired status."""
         now = 1500000000
         field_invite = DocumentGroupV2FieldInvite(
@@ -83,7 +82,7 @@ def test_simplified_invite_participant_from_field_invite_expired_status(self):
             expiration_time=2000000000,
             expiration_days=30,
             signer_email="test@example.com",
-            password_protected="0",
+            password_protected="0",  # noqa: S106
             email_statuses=[],
         )
         participant = SimplifiedInviteParticipant.from_document_group_v2_field_invite(field_invite, now)
@@ -99,7 +98,7 @@ def test_simplified_invite_participant_from_field_invite_expired_status(self):
             ("signed", 2000000000, 2500000000, False),  # Past expiration but signed status (not in PENDING)
         ],
     )
-    def test_check_expired(self, status: str, expires_at: int | None, now: int, expected: bool):
+    def test_check_expired(self, status: str, expires_at: int | None, now: int, expected: bool) -> None:
         """Test check_expired method with various status, expiration, and time combinations."""
         result = SimplifiedInviteParticipant.check_expired(status, expires_at, now)
         assert result == expected