refactor: reorg test files

Author: gao-sun

mcpauth/__init__.py

@@ -1,5 +1,8 @@
 import logging
+from typing import Any, Literal, Union
 
+from .middleware.create_bearer_auth import BaseBearerAuthConfig, BearerAuthConfig
+from .types import VerifyAccessTokenFunction
 from .utils.fetch_server_config import ServerMetadataPaths
 from .config import MCPAuthConfig
 from .exceptions import MCPAuthAuthServerException, AuthServerExceptionCode
@@ -14,6 +17,10 @@ def __init__(self, config: MCPAuthConfig):
         result = validate_server_config(config.server)
 
         if not result.is_valid:
+            logging.error(
+                "The authorization server configuration is invalid:\n"
+                f"{result.errors}\n"
+            )
             raise MCPAuthAuthServerException(
                 AuthServerExceptionCode.INVALID_SERVER_CONFIG, cause=result
             )
@@ -55,3 +62,47 @@ async def dispatch(
                     return await call_next(request)
 
         return DelegatedMiddleware
+
+    def bearer_auth_middleware(
+        self,
+        mode_or_verify: Union[Literal["jwt"], VerifyAccessTokenFunction],
+        config: BaseBearerAuthConfig,
+        jwt_options: dict[str, Any] = {},
+    ) -> type[BaseHTTPMiddleware]:
+        """
+        Creates a middleware that handles bearer token authentication.
+
+        :param mode_or_verify: If "jwt", uses built-in JWT verification; or a custom function that
+        takes a string token and returns an `AuthInfo` object.
+        :param config: Configuration for the Bearer auth handler, including audience, required
+        scopes, etc.
+        :param jwt_options: Optional dictionary of additional options for JWT verification
+        (`jwt.decode`). Not used if a custom function is provided.
+        :return: A middleware class that can be used in a Starlette or FastAPI application.
+        """
+
+        metadata = self.config.server.metadata
+        if isinstance(mode_or_verify, str) and mode_or_verify == "jwt":
+            from .utils.create_verify_jwt import create_verify_jwt
+
+            if not metadata.jwks_uri:
+                raise MCPAuthAuthServerException(
+                    AuthServerExceptionCode.MISSING_JWKS_URI
+                )
+
+            verify = create_verify_jwt(
+                metadata.jwks_uri,
+                options=jwt_options,
+            )
+        elif callable(mode_or_verify):
+            verify = mode_or_verify
+        else:
+            raise ValueError(
+                "mode_or_verify must be 'jwt' or a callable function that verifies tokens."
+            )
+
+        from .middleware.create_bearer_auth import create_bearer_auth
+
+        return create_bearer_auth(
+            verify, BearerAuthConfig(issuer=metadata.issuer, **config.model_dump())
+        )

mcpauth/middleware/create_bearer_auth.py

@@ -18,21 +18,39 @@
 from ..types import VerifyAccessTokenFunction, Record
 
 
-class BearerAuthConfig(BaseModel):
+class BaseBearerAuthConfig(BaseModel):
     """
-    Configuration for the Bearer auth handler.
-
-    Attributes:
-      issuer: The expected issuer of the access token.
-      audience: The expected audience of the access token.
-      required_scopes: An array of required scopes that the access token must have.
-      show_error_details: Whether to show detailed error information in the response.
+    Base configuration for the Bearer auth handler.
     """
 
-    issuer: str
     audience: Optional[str] = None
+    """
+    The expected audience of the access token. If not provided, no audience check is performed.
+    """
+
     required_scopes: Optional[List[str]] = None
+    """
+    An array of required scopes that the access token must have. If not provided, no scope check is
+    performed.
+    """
+
     show_error_details: bool = False
+    """
+    Whether to show detailed error information in the response. Defaults to False.
+    If True, detailed error information will be included in the response body for debugging
+    purposes.
+    """
+
+
+class BearerAuthConfig(BaseBearerAuthConfig):
+    """
+    Configuration for the Bearer auth handler.
+    """
+
+    issuer: str
+    """
+    The expected issuer of the access token. This should be a valid URL.
+    """
 
 
 def get_bearer_token_from_headers(headers: Headers) -> str:

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+pythonpath = .

mcpauth/__init__test.py renamed to tests/__init__test.py

@@ -6,6 +6,10 @@
 from mcpauth.config import MCPAuthConfig
 from mcpauth.models.auth_server import AuthServerConfig, AuthServerType
 from mcpauth.models.oauth import AuthorizationServerMetadata
+from mcpauth.middleware.create_bearer_auth import BaseBearerAuthConfig
+from mcpauth.middleware.create_bearer_auth import BaseBearerAuthConfig
+from mcpauth.middleware.create_bearer_auth import BaseBearerAuthConfig
+from mcpauth.middleware.create_bearer_auth import BaseBearerAuthConfig
 
 
 class TestMCPAuth:
@@ -72,6 +76,8 @@ def test_init_with_warnings(self, mock_warning: MagicMock):
         # Verify
         assert mock_warning.called
 
+
+class TestDelegatedMiddleware:
     @pytest.mark.asyncio
     async def test_delegated_middleware_oauth_endpoint(self):
         # Setup
@@ -135,3 +141,122 @@ async def test_delegated_middleware_other_endpoint(self):
         # Verify
         assert mock_call_next.called
         assert response == mock_response
+
+
+class TestBearerAuthMiddleware:
+    def test_bearer_auth_middleware_jwt_mode(self):
+        # Setup
+        server_config = AuthServerConfig(
+            type=AuthServerType.OAUTH,
+            metadata=AuthorizationServerMetadata(
+                issuer="https://example.com",
+                authorization_endpoint="https://example.com/oauth/authorize",
+                token_endpoint="https://example.com/oauth/token",
+                jwks_uri="https://example.com/.well-known/jwks.json",
+                response_types_supported=["code"],
+                grant_types_supported=["authorization_code"],
+                code_challenge_methods_supported=["S256"],
+            ),
+        )
+        config = MCPAuthConfig(server=server_config)
+        auth = MCPAuth(config)
+
+        # Exercise
+        with patch(
+            "mcpauth.utils.create_verify_jwt.create_verify_jwt"
+        ) as mock_create_verify_jwt:
+            mock_create_verify_jwt.return_value = MagicMock()
+            middleware_class = auth.bearer_auth_middleware(
+                "jwt", BaseBearerAuthConfig(required_scopes=["profile"])
+            )
+
+        # Verify
+        assert middleware_class is not None
+        mock_create_verify_jwt.assert_called_once_with(
+            "https://example.com/.well-known/jwks.json", options={}
+        )
+
+    def test_bearer_auth_middleware_custom_verify(self):
+        # Setup
+        server_config = AuthServerConfig(
+            type=AuthServerType.OAUTH,
+            metadata=AuthorizationServerMetadata(
+                issuer="https://example.com",
+                authorization_endpoint="https://example.com/oauth/authorize",
+                token_endpoint="https://example.com/oauth/token",
+                response_types_supported=["code"],
+                grant_types_supported=["authorization_code"],
+                code_challenge_methods_supported=["S256"],
+            ),
+        )
+        config = MCPAuthConfig(server=server_config)
+        auth = MCPAuth(config)
+
+        custom_verify = MagicMock()
+
+        # Exercise
+        with patch(
+            "mcpauth.middleware.create_bearer_auth.create_bearer_auth"
+        ) as mock_create_bearer_auth:
+            middleware_class = auth.bearer_auth_middleware(
+                custom_verify, BaseBearerAuthConfig(required_scopes=["profile"])
+            )
+
+        # Verify
+        assert middleware_class is not None
+        mock_create_bearer_auth.assert_called_once()
+        args, kwargs = mock_create_bearer_auth.call_args
+        assert args[0] == custom_verify
+        assert kwargs == {}
+
+    def test_bearer_auth_middleware_jwt_without_jwks_uri(self):
+        # Setup
+        server_config = AuthServerConfig(
+            type=AuthServerType.OAUTH,
+            metadata=AuthorizationServerMetadata(
+                issuer="https://example.com",
+                authorization_endpoint="https://example.com/oauth/authorize",
+                token_endpoint="https://example.com/oauth/token",
+                # No jwks_uri
+                response_types_supported=["code"],
+                grant_types_supported=["authorization_code"],
+                code_challenge_methods_supported=["S256"],
+            ),
+        )
+        config = MCPAuthConfig(server=server_config)
+        auth = MCPAuth(config)
+
+        # Exercise & Verify
+        with pytest.raises(MCPAuthAuthServerException) as exc_info:
+            auth.bearer_auth_middleware(
+                "jwt", BaseBearerAuthConfig(required_scopes=["profile"])
+            )
+
+        assert exc_info.value.code == AuthServerExceptionCode.MISSING_JWKS_URI
+
+    def test_bearer_auth_middleware_invalid_mode(self):
+        # Setup
+        server_config = AuthServerConfig(
+            type=AuthServerType.OAUTH,
+            metadata=AuthorizationServerMetadata(
+                issuer="https://example.com",
+                authorization_endpoint="https://example.com/oauth/authorize",
+                token_endpoint="https://example.com/oauth/token",
+                response_types_supported=["code"],
+                grant_types_supported=["authorization_code"],
+                code_challenge_methods_supported=["S256"],
+            ),
+        )
+        config = MCPAuthConfig(server=server_config)
+        auth = MCPAuth(config)
+
+        # Exercise & Verify
+        with pytest.raises(ValueError) as exc_info:
+            auth.bearer_auth_middleware(
+                "invalid_mode",  # type: ignore
+                BaseBearerAuthConfig(required_scopes=["profile"]),
+            )
+
+        assert "mode_or_verify must be 'jwt' or a callable function" in str(
+            exc_info.value
+        )