fix(keycardai-mcp): support x-forwarded-port header

Author: kamil-keycard

packages/mcp/src/keycardai/mcp/server/handlers/metadata.py

@@ -10,6 +10,8 @@
 
 from keycardai.oauth.types.oauth import GrantType, TokenEndpointAuthMethod
 
+from ..shared.starlette import get_base_url
+
 
 class InferredProtectedResourceMetadata(ProtectedResourceMetadata):
     """Extended ProtectedResourceMetadata that allows resource to be inferred from request."""
@@ -56,6 +58,7 @@ def _strip_zone_id_from_path(zone_id: str, path: str) -> str:
         return path[len(zone_id):]
     return path
 
+
 def _create_resource_url(base_url: str | AnyHttpUrl, path: str) -> AnyHttpUrl:
     base_url_str = str(base_url).rstrip("/")
     if path and not path.startswith("/"):
@@ -80,13 +83,17 @@ def wrapper(request: Request) -> Response:
         # Create a copy of the metadata to avoid mutating the original
         request_metadata = metadata.model_copy(deep=True)
         path = _remove_well_known_prefix(request.url.path)
+
+        # Get proxy-aware base URL for correct scheme handling
+        base_url = get_base_url(request)
+
         if enable_multi_zone:
             zone_id = _get_zone_id_from_path(path)
             if zone_id:
                 request_metadata.authorization_servers = [ _create_zone_scoped_authorization_server_url(zone_id, request_metadata.authorization_servers[0]) ]
 
-        request_metadata.resource = _create_resource_url(request.base_url, path)
-        request_metadata.jwks_uri = _create_jwks_uri(str(request.base_url))
+        request_metadata.resource = _create_resource_url(base_url, path)
+        request_metadata.jwks_uri = _create_jwks_uri(base_url)
         request_metadata.client_id = str(request_metadata.resource)
         request_metadata.client_name = "MCP Server"
         request_metadata.token_endpoint_auth_method = TokenEndpointAuthMethod.PRIVATE_KEY_JWT
@@ -96,7 +103,7 @@ def wrapper(request: Request) -> Response:
         mcp_version = request.headers.get("mcp-protocol-version")
         # TODO: what is the reason for this?
         if mcp_version == "2025-03-26":
-            json["authorization_servers"] = [ request.base_url ]
+            json["authorization_servers"] = [ base_url ]
         return Response(content=request_metadata.model_dump_json(exclude_none=True), status_code=200)
     return wrapper
 
@@ -115,7 +122,8 @@ def wrapper(request: Request) -> Response:
                 resp = client.get(f"{actual_issuer}/.well-known/oauth-authorization-server")
                 resp.raise_for_status()
                 authorization_server_metadata = resp.json()
-                authorization_server_metadata["authorization_endpoint"] = f"{request.base_url}{authorization_server_metadata['authorization_endpoint']}"
+                base_url = get_base_url(request)
+                authorization_server_metadata["authorization_endpoint"] = f"{base_url}{authorization_server_metadata['authorization_endpoint']}"
                 return Response(content=json.dumps(authorization_server_metadata), status_code=200)
         except httpx.HTTPStatusError as e:
             # Return the same status code as the upstream server

packages/mcp/src/keycardai/mcp/server/middleware/bearer.py

@@ -7,11 +7,12 @@
 from starlette.types import ASGIApp
 
 from ..auth.verifier import TokenVerifier
+from ..shared.starlette import get_base_url
 
 
 def _get_oauth_protected_resource_url(request: Request) -> str:
     path = request.url.path.lstrip("/").rstrip("/")
-    base_url = str(request.base_url).rstrip("/")
+    base_url = get_base_url(request)
     return str(AnyHttpUrl(f"{base_url}/.well-known/oauth-protected-resource/{path}"))
 
 def _get_bearer_token(request: Request) -> str | None:

packages/mcp/src/keycardai/mcp/server/shared/starlette.py

@@ -0,0 +1,22 @@
+"""Shared utilities for Starlette/FastAPI applications."""
+
+from pydantic import AnyHttpUrl
+from starlette.requests import Request
+
+"""Supported protocols for the base URL."""
+SUPPORTED_PROTOCOLS = ["http", "https"]
+
+
+def get_base_url(request: Request) -> str:
+    """Get the correct base URL considering proxy headers like X-Forwarded-Proto."""
+    request_base_url = AnyHttpUrl(str(request.base_url))
+    proto = request.headers.get("x-forwarded-proto") or request_base_url.scheme
+    if proto not in SUPPORTED_PROTOCOLS:
+        proto = "https"
+
+    if request_base_url.port not in [443, 80]:
+        base_url = f"{proto}://{request_base_url.host}:{request_base_url.port}"
+    else:
+        base_url = f"{proto}://{request_base_url.host}"
+
+    return base_url

packages/mcp/tests/keycardai/mcp/server/handlers/test_metadata.py

@@ -5,6 +5,8 @@
 
 import pytest
 from pydantic import AnyHttpUrl
+from starlette.datastructures import URL
+from starlette.requests import Request
 
 from keycardai.mcp.server.handlers.metadata import (
     _create_resource_url,
@@ -14,6 +16,7 @@
     _remove_well_known_prefix,
     _strip_zone_id_from_path,
 )
+from keycardai.mcp.server.shared.starlette import get_base_url
 
 
 class TestIsAuthorizationServerZoneScoped:
@@ -399,5 +402,86 @@ def test_case_sensitivity(self):
         assert result == "zone123/api/v1"
 
 
+class TestGetBaseUrl:
+    """Test get_base_url function."""
+
+    def _create_mock_request(self, base_url: str, headers: dict[str, str] | None = None) -> Request:
+        """Create a mock request with specified base URL and headers."""
+        if headers is None:
+            headers = {}
+
+        parsed_url = URL(base_url)
+        # Create a minimal ASGI scope for testing
+        scope = {
+            "type": "http",
+            "method": "GET",
+            "scheme": parsed_url.scheme,
+            "server": (parsed_url.hostname, parsed_url.port or (443 if parsed_url.scheme == "https" else 80)),
+            "path": parsed_url.path or "/",
+            "query_string": b"",
+            "headers": [(k.lower().encode(), v.encode()) for k, v in headers.items()],
+        }
+        return Request(scope)
+
+    def test_no_proxy_headers(self):
+        """Test with no proxy headers - should return original base URL."""
+        request = self._create_mock_request("http://example.com")
+        result = get_base_url(request)
+        assert result == "http://example.com"
+
+    def test_with_x_forwarded_proto_https(self):
+        """Test with X-Forwarded-Proto header indicating HTTPS."""
+        headers = {"x-forwarded-proto": "https"}
+        request = self._create_mock_request("http://example.com", headers)
+        result = get_base_url(request)
+        assert result == "https://example.com"
+
+    def test_with_x_forwarded_proto_http(self):
+        """Test with X-Forwarded-Proto header indicating HTTP."""
+        headers = {"x-forwarded-proto": "http"}
+        request = self._create_mock_request("https://example.com", headers)
+        result = get_base_url(request)
+        assert result == "http://example.com"
+
+    def test_with_port_number(self):
+        """Test with port number in base URL."""
+        headers = {"x-forwarded-proto": "https"}
+        request = self._create_mock_request("http://example.com:8080", headers)
+        result = get_base_url(request)
+        assert result == "https://example.com:8080"
+
+    def test_with_path_in_base_url(self):
+        """Test with path in base URL - path should be ignored for base URL."""
+        headers = {"x-forwarded-proto": "https"}
+        request = self._create_mock_request("http://example.com/api/v1", headers)
+        result = get_base_url(request)
+        assert result == "https://example.com"
+
+    def test_case_insensitive_header(self):
+        """Test that header matching is case insensitive (Starlette handles this)."""
+        headers = {"X-Forwarded-Proto": "https"}
+        request = self._create_mock_request("http://example.com", headers)
+        result = get_base_url(request)
+        assert result == "https://example.com"
+
+    def test_trailing_slash_handling(self):
+        """Test that trailing slashes are properly handled."""
+        headers = {"x-forwarded-proto": "https"}
+        request = self._create_mock_request("http://example.com/", headers)
+        result = get_base_url(request)
+        assert result == "https://example.com"
+
+    def test_aws_app_runner_scenario(self):
+        """Test the specific AWS App Runner scenario from the issue."""
+        headers = {
+            "host": "ppxrhd2bw4.us-east-1.awsapprunner.com",
+            "x-forwarded-proto": "https",
+            "x-forwarded-for": "92.238.31.228"
+        }
+        request = self._create_mock_request("http://ppxrhd2bw4.us-east-1.awsapprunner.com", headers)
+        result = get_base_url(request)
+        assert result == "https://ppxrhd2bw4.us-east-1.awsapprunner.com"
+
+
 if __name__ == "__main__":
     pytest.main([__file__])

packages/mcp/tests/keycardai/mcp/server/middleware/test_bearer.py

@@ -246,3 +246,51 @@ def test_case_sensitivity_of_auth_scheme(self):
 
             result = _get_bearer_token(request)
             assert result == expected_token, f"Failed for auth scheme: {auth_scheme}"
+
+
+class TestGetBaseUrlMiddleware:
+    """Tests for get_base_url function in middleware."""
+
+    def _create_mock_request(self, base_url: str, path: str = "/", headers: dict[str, str] | None = None) -> Request:
+        """Create a mock request with specified base URL, path, and headers."""
+        if headers is None:
+            headers = {}
+
+        # Create a minimal ASGI scope for testing
+        scope = {
+            "type": "http",
+            "method": "GET",
+            "scheme": URL(base_url).scheme,
+            "server": (URL(base_url).hostname, URL(base_url).port or (443 if URL(base_url).scheme == "https" else 80)),
+            "path": path,
+            "query_string": b"",
+            "headers": [(k.lower().encode(), v.encode()) for k, v in headers.items()],
+        }
+        return Request(scope)
+
+    def test_proxy_aware_url_in_oauth_resource_url(self):
+        """Test that _get_oauth_protected_resource_url uses proxy-aware base URL."""
+        headers = {"x-forwarded-proto": "https"}
+        request = self._create_mock_request("http://example.com", "/api/resource", headers)
+
+        result = _get_oauth_protected_resource_url(request)
+        assert result == "https://example.com/.well-known/oauth-protected-resource/api/resource"
+
+    def test_no_proxy_headers_in_oauth_resource_url(self):
+        """Test _get_oauth_protected_resource_url without proxy headers."""
+        request = self._create_mock_request("http://example.com", "/api/resource")
+
+        result = _get_oauth_protected_resource_url(request)
+        assert result == "http://example.com/.well-known/oauth-protected-resource/api/resource"
+
+    def test_aws_app_runner_scenario_in_middleware(self):
+        """Test the AWS App Runner scenario in middleware context."""
+        headers = {
+            "host": "ppxrhd2bw4.us-east-1.awsapprunner.com",
+            "x-forwarded-proto": "https",
+            "x-forwarded-for": "92.238.31.228"
+        }
+        request = self._create_mock_request("http://ppxrhd2bw4.us-east-1.awsapprunner.com", "/zone123/api", headers)
+
+        result = _get_oauth_protected_resource_url(request)
+        assert result == "https://ppxrhd2bw4.us-east-1.awsapprunner.com/.well-known/oauth-protected-resource/zone123/api"