feat: thread mTLS ssl_context and alias routing through MFA verify

Author: cschetan77

src/auth0_server_python/auth_server/mfa_client.py

@@ -4,6 +4,7 @@
 """
 
 import json
+import ssl
 import time
 from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Any, Optional, Union
@@ -74,6 +75,8 @@ def __init__(
         ] = None,
         mfa_token_ttl: int = DEFAULT_MFA_TOKEN_TTL,
         apply_client_authentication: Optional[Callable] = None,
+        use_mtls: bool = False,
+        ssl_context: Optional[ssl.SSLContext] = None,
     ):
         if callable(domain):
             self._domain = None
@@ -92,10 +95,14 @@ def __init__(
             raise ConfigurationError("mfa_token_ttl must be a positive number of seconds")
         self._mfa_token_ttl = mfa_token_ttl
         self._apply_client_authentication = apply_client_authentication
+        self._use_mtls = use_mtls
+        self._ssl_context = ssl_context
 
     def _get_http_client(self, **kwargs) -> httpx.AsyncClient:
         """Return an httpx.AsyncClient with default headers injected."""
         headers = {**kwargs.pop("headers", {}), **self._headers}
+        if self._use_mtls and "verify" not in kwargs:
+            kwargs["verify"] = self._ssl_context
         return httpx.AsyncClient(headers=headers, **kwargs)
 
     def _apply_mfa_client_authentication(self, body: dict, base_url: str) -> None:
@@ -472,6 +479,7 @@ async def verify(
         options: dict[str, Any],
         store_options: Optional[dict[str, Any]] = None,
         dpop_key: Optional["jwk.JWK"] = None,
+        token_endpoint_override: Optional[str] = None,
     ) -> MfaVerifyResponse:
         """
         Verifies an MFA code and completes authentication.
@@ -504,6 +512,12 @@ async def verify(
             MfaRequiredError: When chained MFA is required.
             ConfigurationError: If neither client_secret nor client_assertion_signing_key is configured.
         """
+        if self._use_mtls and dpop_key is not None:
+            raise ConfigurationError(
+                "dpop_key cannot be combined with use_mtls. DPoP and mTLS bind tokens "
+                "differently; DPoP would take precedence and the token would not be "
+                "certificate-bound."
+            )
         mfa_token = options.get("mfa_token")
         if not mfa_token:
             raise MfaTokenInvalidError()
@@ -534,7 +548,7 @@ async def verify(
             )
 
         try:
-            token_endpoint = f"{base_url}/oauth/token"
+            token_endpoint = token_endpoint_override or f"{base_url}/oauth/token"
 
             async with self._get_http_client() as client:
                 headers = {"Content-Type": "application/x-www-form-urlencoded"}

src/auth0_server_python/auth_server/server_client.py

@@ -265,6 +265,8 @@ def __init__(
             session_establisher=self._establish_session_from_mfa_verify_response,
             mfa_token_ttl=mfa_token_ttl,
             apply_client_authentication=self._apply_client_authentication,
+            use_mtls=self._use_mtls,
+            ssl_context=self._ssl_context,
         )
 
         self._passwordless_client = PasswordlessClient(self)

src/auth0_server_python/tests/test_mfa_client.py

@@ -3,6 +3,7 @@
 """
 
 import json
+import ssl
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -1093,3 +1094,58 @@ async def mock_post(self_client, url, **kwargs):
         result = await client.verify({"mfa_token": _enc(), "otp": "123456"})
         assert result.token_type == "Bearer"
         assert "DPoP" not in captured_request["kwargs"]["headers"]
+
+
+# ============================================================================
+# mTLS — MfaClient SSLContext threading + DPoP exclusion + endpoint override
+# ============================================================================
+
+
+def _mtls_mfa_client() -> MfaClient:
+    return MfaClient(
+        domain=DOMAIN,
+        client_id=CLIENT_ID,
+        client_secret=None,
+        secret=SECRET,
+        use_mtls=True,
+        ssl_context=ssl.create_default_context(),
+    )
+
+
+@pytest.mark.asyncio
+async def test_mfa_get_http_client_passes_ssl_context(mocker):
+    mfa = _mtls_mfa_client()
+    spy = mocker.patch("auth0_server_python.auth_server.mfa_client.httpx.AsyncClient")
+    mfa._get_http_client()
+    _, kwargs = spy.call_args
+    assert kwargs.get("verify") is mfa._ssl_context
+
+
+@pytest.mark.asyncio
+async def test_mfa_verify_rejects_dpop_under_mtls():
+    mfa = _mtls_mfa_client()
+    with pytest.raises(ConfigurationError):
+        await mfa.verify({"mfa_token": _enc(), "otp": "123456"}, dpop_key=object())
+
+
+@pytest.mark.asyncio
+async def test_mfa_verify_uses_token_endpoint_override(mocker):
+    mfa = _mtls_mfa_client()
+    response = AsyncMock()
+    response.status_code = 200
+    response.json = MagicMock(return_value={
+        "access_token": "at", "token_type": "Bearer", "expires_in": 3600
+    })
+    captured = {}
+
+    async def mock_post(self_client, url, **kwargs):
+        captured["url"] = url
+        return response
+
+    mocker.patch("httpx.AsyncClient.post", new=mock_post)
+
+    await mfa.verify(
+        {"mfa_token": _enc(), "otp": "123456"},
+        token_endpoint_override="https://mtls.auth0.local/oauth/token",
+    )
+    assert captured["url"] == "https://mtls.auth0.local/oauth/token"