Update JWT login type to support JWKS, custom sub claim, and encode special chars in user ID

changelog.d/9493.feature

@@ -0,0 +1 @@
+Update JWT login type to support JSON Web Key Sets (JWKS), custom sub claim, and option to encode unsupported characters in user ID.

docs/sample_config.yaml

@@ -2158,10 +2158,30 @@ sso:
     # This is either the private shared secret or the public key used to
     # decode the contents of the JSON web token.
     #
-    # Required if 'enabled' is true.
+    # Either 'secret' or 'jwks_uri' is required if 'enabled' is true.
     #
     #secret: "provided-by-your-issuer"
 
+    # URI where to fetch the JWKS containing the public keys that
+    # should be used to verify the signature of the JSON web token.
+    # Only used if 'secret' is not provided.
+    #
+    # Either 'secret' or 'jwks_uri' is required if 'enabled' is true.
+    #
+    #jwks_uri: "provided-by-your-issuer"
+
+    # Name of the claim containing a unique identifier for the user.
+    #
+    # Optional, defaults to `sub`.
+    #
+    #subject_claim: "sub"
+
+    # Perform normalization of the user ID and encode unsupported characters.
+    #
+    # Optional, defaults to false.
+    #
+    #normalize_user_id: true
+
     # The algorithm used to sign the JSON web token.
     #
     # Supported algorithms are listed at

synapse/config/jwt_config.py

@@ -29,11 +29,14 @@ def read_config(self, config, **kwargs):
         jwt_config = config.get("jwt_config", None)
         if jwt_config:
             self.jwt_enabled = jwt_config.get("enabled", False)
-            self.jwt_secret = jwt_config["secret"]
             self.jwt_algorithm = jwt_config["algorithm"]
 
             # The issuer and audiences are optional, if provided, it is asserted
             # that the claims exist on the JWT.
+            self.jwt_secret = jwt_config.get("secret")
+            self.jwt_jwks_uri = jwt_config.get("jwks_uri")
+            self.jwt_subject_claim = jwt_config.get("subject_claim")
+            self.jwt_normalize_user_id = jwt_config.get("normalize_user_id", False)
             self.jwt_issuer = jwt_config.get("issuer")
             self.jwt_audiences = jwt_config.get("audiences")
 
@@ -46,6 +49,9 @@ def read_config(self, config, **kwargs):
         else:
             self.jwt_enabled = False
             self.jwt_secret = None
+            self.jwt_jwks_uri = None
+            self.jwt_subject_claim = None
+            self.jwt_normalize_user_id = False
             self.jwt_algorithm = None
             self.jwt_issuer = None
             self.jwt_audiences = None
@@ -76,10 +82,30 @@ def generate_config_section(self, **kwargs):
             # This is either the private shared secret or the public key used to
             # decode the contents of the JSON web token.
             #
-            # Required if 'enabled' is true.
+            # Either 'secret' or 'jwks_uri' is required if 'enabled' is true.
             #
             #secret: "provided-by-your-issuer"
 
+            # URI where to fetch the JWKS containing the public keys that
+            # should be used to verify the signature of the JSON web token.
+            # Only used if 'secret' is not provided.
+            #
+            # Either 'secret' or 'jwks_uri' is required if 'enabled' is true.
+            #
+            #jwks_uri: "provided-by-your-issuer"
+
+            # Name of the claim containing a unique identifier for the user.
+            #
+            # Optional, defaults to `sub`.
+            #
+            #subject_claim: "sub"
+
+            # Perform normalization of the user ID and encode unsupported characters.
+            #
+            # Optional, defaults to false.
+            #
+            #normalize_user_id: true
+
             # The algorithm used to sign the JSON web token.
             #
             # Supported algorithms are listed at

synapse/rest/client/v1/login.py

@@ -29,7 +29,7 @@
 from synapse.http.site import SynapseRequest
 from synapse.rest.client.v2_alpha._base import client_patterns
 from synapse.rest.well_known import WellKnownBuilder
-from synapse.types import JsonDict, UserID
+from synapse.types import JsonDict, UserID, map_username_to_mxid_localpart
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -53,6 +53,9 @@ def __init__(self, hs: "HomeServer"):
         # JWT configuration variables.
         self.jwt_enabled = hs.config.jwt_enabled
         self.jwt_secret = hs.config.jwt_secret
+        self.jwt_jwks_uri = hs.config.jwt_jwks_uri
+        self.jwt_subject_claim = hs.config.jwt_subject_claim
+        self.jwt_normalize_user_id = hs.config.jwt_normalize_user_id
         self.jwt_algorithm = hs.config.jwt_algorithm
         self.jwt_issuer = hs.config.jwt_issuer
         self.jwt_audiences = hs.config.jwt_audiences
@@ -298,11 +301,19 @@ async def _do_jwt_login(self, login_submission: JsonDict) -> Dict[str, str]:
             )
 
         import jwt
+        from jwt import PyJWKClient
+
+        key = self.jwt_secret
+
+        if not key and self.jwt_jwks_uri:
+            jwks_client = PyJWKClient(self.jwt_jwks_uri)
+            signing_key = jwks_client.get_signing_key_from_jwt(token)
+            key = signing_key.key
 
         try:
             payload = jwt.decode(
                 token,
-                self.jwt_secret,
+                key,
                 algorithms=[self.jwt_algorithm],
                 issuer=self.jwt_issuer,
                 audience=self.jwt_audiences,
@@ -315,10 +326,14 @@ async def _do_jwt_login(self, login_submission: JsonDict) -> Dict[str, str]:
                 errcode=Codes.FORBIDDEN,
             )
 
-        user = payload.get("sub", None)
+        subject_claim = self.jwt_subject_claim or "sub"
+        user = payload.get(subject_claim, None)
         if user is None:
             raise LoginError(403, "Invalid JWT", errcode=Codes.FORBIDDEN)
 
+        if self.jwt_normalize_user_id:
+            user = map_username_to_mxid_localpart(user)
+
         user_id = UserID(user, self.hs.hostname).to_string()
         result = await self._complete_login(
             user_id, login_submission, create_non_existent_users=True

tests/rest/client/v1/test_login.py

@@ -982,6 +982,54 @@ def test_login_aud_no_config(self):
             channel.json_body["error"], "JWT validation failed: Invalid audience"
         )
 
+    def test_login_default_sub(self):
+        """Test reading user ID from the default subject claim."""
+        channel = self.jwt_login({"sub": "kermit"})
+        self.assertEqual(channel.result["code"], b"200", channel.result)
+        self.assertEqual(channel.json_body["user_id"], "@kermit:test")
+
+    @override_config(
+        {
+            "jwt_config": {
+                "jwt_enabled": True,
+                "secret": jwt_secret,
+                "algorithm": jwt_algorithm,
+                "subject_claim": "username",
+            }
+        }
+    )
+    def test_login_custom_sub(self):
+        """Test reading user ID from a custom subject claim."""
+        channel = self.jwt_login({"username": "frog"})
+        self.assertEqual(channel.result["code"], b"200", channel.result)
+        self.assertEqual(channel.json_body["user_id"], "@frog:test")
+
+    def test_login_no_normalize_id(self):
+        """Test mapping user ID to Matrix ID without normalization"""
+        channel = self.jwt_login({"sub": "#kermit"})
+        self.assertEqual(channel.result["code"], b"400", channel.result)
+        self.assertEqual(channel.json_body["errcode"], "M_INVALID_USERNAME")
+        self.assertEqual(
+            channel.json_body["error"],
+            "User ID can only contain characters a-z, 0-9, or '=_-./'",
+        )
+
+    @override_config(
+        {
+            "jwt_config": {
+                "jwt_enabled": True,
+                "secret": jwt_secret,
+                "algorithm": jwt_algorithm,
+                "normalize_user_id": True,
+            }
+        }
+    )
+    def test_login_normalize_id(self):
+        """Test mapping user ID to Matrix ID with normalization"""
+        channel = self.jwt_login({"sub": "#kermit"})
+        self.assertEqual(channel.result["code"], b"200", channel.result)
+        self.assertEqual(channel.json_body["user_id"], "@=23kermit:test")
+
     def test_login_no_token(self):
         params = {"type": "org.matrix.login.jwt"}
         channel = self.make_request(b"POST", LOGIN_URL, params)