feat: warn when mTLS token lacks cnf.x5t#S256 binding

Author: cschetan77

src/auth0_server_python/auth_server/server_client.py

@@ -7,6 +7,7 @@
 import json
 import ssl
 import time
+import warnings
 from collections import OrderedDict
 from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union
 
@@ -441,6 +442,31 @@ async def _verify_and_decode_jwt(
 
         return jwt.decode(token, signing_key.key, **kwargs)
 
+    def _warn_if_not_cert_bound(self, access_token: Optional[str]) -> None:
+        """Advisory warning when mTLS is on but the access token is not certificate-bound.
+
+        Silent on opaque (non-JWT) tokens and when mTLS is off; never raises.
+        """
+        if not self._use_mtls or not access_token:
+            return
+        try:
+            claims = jwt.decode(
+                access_token,
+                options={"verify_signature": False},
+                algorithms=["HS256", "RS256", "ES256", "PS256"],
+            )
+        except Exception:
+            return  # opaque or unparseable token — nothing to assert
+        cnf = claims.get("cnf") if isinstance(claims, dict) else None
+        if not (isinstance(cnf, dict) and cnf.get("x5t#S256")):
+            warnings.warn(
+                "mTLS is enabled but the access token is not certificate-bound "
+                "(no cnf.x5t#S256). Sender-constraining is not active — configure "
+                "Token Sender-Constraining (mTLS) on the API resource server.",
+                UserWarning,
+                stacklevel=2,
+            )
+
     async def _fetch_oidc_metadata(self, domain: str) -> dict:
         """Fetch OIDC metadata from domain."""
         normalized_domain = self._normalize_url(domain)
@@ -813,6 +839,8 @@ async def complete_interactive_login(
             raise ApiError(
                 "token_error", f"Token exchange failed: {str(e)}", e)
 
+        self._warn_if_not_cert_bound(token_response.get("access_token"))
+
         # Use the userinfo field from the token_response for user claims
         user_info = token_response.get("userinfo")
         user_claims = None
@@ -1491,6 +1519,8 @@ async def get_token_by_refresh_token(self, options: dict[str, Any]) -> dict[str,
 
                 token_response = response.json()
 
+                self._warn_if_not_cert_bound(token_response.get("access_token"))
+
                 # Add required fields if they are missing
                 if "expires_in" in token_response and "expires_at" not in token_response:
                     token_response["expires_at"] = int(

src/auth0_server_python/tests/test_server_client.py

@@ -9805,6 +9805,47 @@ async def test_apply_client_auth_mtls_returns_none_and_strips_creds():
     assert "client_assertion_type" not in params
 
 
+_TEST_JWT_KEY = "test-signing-key-for-mtls-tests-32b"  # ≥32 bytes avoids InsecureKeyLengthWarning
+
+
+@pytest.mark.asyncio
+async def test_warn_when_jwt_missing_cnf_under_mtls(recwarn):
+    client = _mtls_client()
+    token = jwt.encode({"sub": "u", "aud": "api"}, _TEST_JWT_KEY, algorithm="HS256")
+    client._warn_if_not_cert_bound(token)
+    assert any(
+        "cnf" in str(w.message).lower() or "certificate-bound" in str(w.message).lower()
+        for w in recwarn.list
+    )
+
+
+@pytest.mark.asyncio
+async def test_no_warn_when_jwt_has_cnf(recwarn):
+    client = _mtls_client()
+    token = jwt.encode({"sub": "u", "cnf": {"x5t#S256": "abc"}}, _TEST_JWT_KEY, algorithm="HS256")
+    client._warn_if_not_cert_bound(token)
+    assert len(recwarn.list) == 0
+
+
+@pytest.mark.asyncio
+async def test_no_warn_on_opaque_token(recwarn):
+    client = _mtls_client()
+    client._warn_if_not_cert_bound("opaque-not-a-jwt")
+    assert len(recwarn.list) == 0
+
+
+@pytest.mark.asyncio
+async def test_warn_never_raises_and_silent_when_not_mtls(recwarn):
+    non_mtls = ServerClient(
+        domain="auth0.local",
+        client_id="<client_id>",
+        client_secret="<client_secret>",
+        secret="<secret>",
+    )
+    non_mtls._warn_if_not_cert_bound(None)
+    assert len(recwarn.list) == 0
+
+
 @pytest.mark.asyncio
 async def test_signin_with_passkey_rejects_dpop_under_mtls(mocker):
     client = _mtls_client()