Do not allow deactivated users to login with JWT

changelog.d/15624.bugfix

@@ -0,0 +1 @@
+Fix a long-standing bug where deactivated users were still able to login using the custom `org.matrix.login.jwt` login type (if enabled).

synapse/handlers/jwt.py

@@ -0,0 +1,118 @@
+# Copyright 2023 Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from typing import TYPE_CHECKING
+
+from authlib.jose import JsonWebToken, JWTClaims
+from authlib.jose.errors import BadSignatureError, InvalidClaimError, JoseError
+
+from synapse.api.errors import Codes, LoginError, StoreError, UserDeactivatedError
+from synapse.types import JsonDict, UserID
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+
+class JwtHandler:
+    def __init__(self, hs: "HomeServer"):
+        self.hs = hs
+        self._main_store = hs.get_datastores().main
+
+        self.jwt_secret = hs.config.jwt.jwt_secret
+        self.jwt_subject_claim = hs.config.jwt.jwt_subject_claim
+        self.jwt_algorithm = hs.config.jwt.jwt_algorithm
+        self.jwt_issuer = hs.config.jwt.jwt_issuer
+        self.jwt_audiences = hs.config.jwt.jwt_audiences
+
+    async def validate_login(self, login_submission: JsonDict) -> str:
+        """
+        Authenticates the user for the /login API
+
+        Args:
+            login_submission: the whole of the login submission
+                (including 'type' and other relevant fields)
+
+        Returns:
+            The user ID that is logging in.
+
+        Raises:
+            LoginError if there was an authentication problem.
+        """
+        token = login_submission.get("token", None)
+        if token is None:
+            raise LoginError(
+                403, "Token field for JWT is missing", errcode=Codes.FORBIDDEN
+            )
+
+        jwt = JsonWebToken([self.jwt_algorithm])
+        claim_options = {}
+        if self.jwt_issuer is not None:
+            claim_options["iss"] = {"value": self.jwt_issuer, "essential": True}
+        if self.jwt_audiences is not None:
+            claim_options["aud"] = {"values": self.jwt_audiences, "essential": True}
+
+        try:
+            claims = jwt.decode(
+                token,
+                key=self.jwt_secret,
+                claims_cls=JWTClaims,
+                claims_options=claim_options,
+            )
+        except BadSignatureError:
+            # We handle this case separately to provide a better error message
+            raise LoginError(
+                403,
+                "JWT validation failed: Signature verification failed",
+                errcode=Codes.FORBIDDEN,
+            )
+        except JoseError as e:
+            # A JWT error occurred, return some info back to the client.
+            raise LoginError(
+                403,
+                "JWT validation failed: %s" % (str(e),),
+                errcode=Codes.FORBIDDEN,
+            )
+
+        try:
+            claims.validate(leeway=120)  # allows 2 min of clock skew
+
+            # Enforce the old behavior which is rolled out in productive
+            # servers: if the JWT contains an 'aud' claim but none is
+            # configured, the login attempt will fail
+            if claims.get("aud") is not None:
+                if self.jwt_audiences is None or len(self.jwt_audiences) == 0:
+                    raise InvalidClaimError("aud")
+        except JoseError as e:
+            raise LoginError(
+                403,
+                "JWT validation failed: %s" % (str(e),),
+                errcode=Codes.FORBIDDEN,
+            )
+
+        user = claims.get(self.jwt_subject_claim, None)
+        if user is None:
+            raise LoginError(403, "Invalid JWT", errcode=Codes.FORBIDDEN)
+
+        user_id = UserID(user, self.hs.hostname).to_string()
+
+        # If the account has been deactivated, do not proceed with the login
+        # flow.
+        try:
+            deactivated = await self._main_store.get_user_deactivated_status(user_id)
+        except StoreError:
+            # JWT lazily creates users, so they may not exist in the database yet.
+            deactivated = False
+        if deactivated:
+            raise UserDeactivatedError("This account has been deactivated")
+
+        return user_id

synapse/rest/client/login.py

@@ -87,11 +87,6 @@ def __init__(self, hs: "HomeServer"):
 
         # JWT configuration variables.
         self.jwt_enabled = hs.config.jwt.jwt_enabled
-        self.jwt_secret = hs.config.jwt.jwt_secret
-        self.jwt_subject_claim = hs.config.jwt.jwt_subject_claim
-        self.jwt_algorithm = hs.config.jwt.jwt_algorithm
-        self.jwt_issuer = hs.config.jwt.jwt_issuer
-        self.jwt_audiences = hs.config.jwt.jwt_audiences
 
         # SSO configuration.
         self.saml2_enabled = hs.config.saml2.saml2_enabled
@@ -427,7 +422,7 @@ async def _do_token_login(
         self, login_submission: JsonDict, should_issue_refresh_token: bool = False
     ) -> LoginResponse:
         """
-        Handle the final stage of SSO login.
+        Handle token login.
 
         Args:
             login_submission: The JSON request body.
@@ -452,72 +447,24 @@ async def _do_token_login(
     async def _do_jwt_login(
         self, login_submission: JsonDict, should_issue_refresh_token: bool = False
     ) -> LoginResponse:
-        token = login_submission.get("token", None)
-        if token is None:
-            raise LoginError(
-                403, "Token field for JWT is missing", errcode=Codes.FORBIDDEN
-            )
-
-        from authlib.jose import JsonWebToken, JWTClaims
-        from authlib.jose.errors import BadSignatureError, InvalidClaimError, JoseError
-
-        jwt = JsonWebToken([self.jwt_algorithm])
-        claim_options = {}
-        if self.jwt_issuer is not None:
-            claim_options["iss"] = {"value": self.jwt_issuer, "essential": True}
-        if self.jwt_audiences is not None:
-            claim_options["aud"] = {"values": self.jwt_audiences, "essential": True}
-
-        try:
-            claims = jwt.decode(
-                token,
-                key=self.jwt_secret,
-                claims_cls=JWTClaims,
-                claims_options=claim_options,
-            )
-        except BadSignatureError:
-            # We handle this case separately to provide a better error message
-            raise LoginError(
-                403,
-                "JWT validation failed: Signature verification failed",
-                errcode=Codes.FORBIDDEN,
-            )
-        except JoseError as e:
-            # A JWT error occurred, return some info back to the client.
-            raise LoginError(
-                403,
-                "JWT validation failed: %s" % (str(e),),
-                errcode=Codes.FORBIDDEN,
-            )
-
-        try:
-            claims.validate(leeway=120)  # allows 2 min of clock skew
-
-            # Enforce the old behavior which is rolled out in productive
-            # servers: if the JWT contains an 'aud' claim but none is
-            # configured, the login attempt will fail
-            if claims.get("aud") is not None:
-                if self.jwt_audiences is None or len(self.jwt_audiences) == 0:
-                    raise InvalidClaimError("aud")
-        except JoseError as e:
-            raise LoginError(
-                403,
-                "JWT validation failed: %s" % (str(e),),
-                errcode=Codes.FORBIDDEN,
-            )
+        """
+        Handle the custom JWT login.
 
-        user = claims.get(self.jwt_subject_claim, None)
-        if user is None:
-            raise LoginError(403, "Invalid JWT", errcode=Codes.FORBIDDEN)
+        Args:
+            login_submission: The JSON request body.
+            should_issue_refresh_token: True if this login should issue
+                a refresh token alongside the access token.
 
-        user_id = UserID(user, self.hs.hostname).to_string()
-        result = await self._complete_login(
+        Returns:
+            The body of the JSON response.
+        """
+        user_id = await self.hs.get_jwt_handler().validate_login(login_submission)
+        return await self._complete_login(
             user_id,
             login_submission,
             create_non_existent_users=True,
             should_issue_refresh_token=should_issue_refresh_token,
         )
-        return result
 
 
 def _get_auth_flow_dict_for_idp(idp: SsoIdentityProvider) -> JsonDict:

synapse/server.py

@@ -147,6 +147,7 @@
 if TYPE_CHECKING:
     from txredisapi import ConnectionHandler
 
+    from synapse.handlers.jwt import JwtHandler
     from synapse.handlers.oidc import OidcHandler
     from synapse.handlers.saml import SamlHandler
 
@@ -533,6 +534,12 @@ def get_typing_handler(self) -> FollowerTypingHandler:
     def get_sso_handler(self) -> SsoHandler:
         return SsoHandler(self)
 
+    @cache_in_self
+    def get_jwt_handler(self) -> "JwtHandler":
+        from synapse.handlers.jwt import JwtHandler
+
+        return JwtHandler(self)
+
     @cache_in_self
     def get_sync_handler(self) -> SyncHandler:
         return SyncHandler(self)

tests/rest/client/test_login.py

@@ -42,7 +42,7 @@
 from tests.unittest import HomeserverTestCase, override_config, skip_unless
 
 try:
-    from authlib.jose import jwk, jwt
+    from authlib.jose import JsonWebKey, jwt
 
     HAS_JWT = True
 except ImportError:
@@ -1054,6 +1054,22 @@ def test_login_no_token(self) -> None:
         self.assertEqual(channel.json_body["errcode"], "M_FORBIDDEN")
         self.assertEqual(channel.json_body["error"], "Token field for JWT is missing")
 
+    def test_deactivated_user(self) -> None:
+        """Logging in as a deactivated account should error."""
+        user_id = self.register_user("kermit", "monkey")
+        self.get_success(
+            self.hs.get_deactivate_account_handler().deactivate_account(
+                user_id, erase_data=False, requester=create_requester(user_id)
+            )
+        )
+
+        channel = self.jwt_login({"sub": "kermit"})
+        self.assertEqual(channel.code, 403, msg=channel.result)
+        self.assertEqual(channel.json_body["errcode"], "M_USER_DEACTIVATED")
+        self.assertEqual(
+            channel.json_body["error"], "This account has been deactivated"
+        )
+
 
 # The JWTPubKeyTestCase is a complement to JWTTestCase where we instead use
 # RSS256, with a public key configured in synapse as "jwt_secret", and tokens
@@ -1121,7 +1137,7 @@ def default_config(self) -> Dict[str, Any]:
     def jwt_encode(self, payload: Dict[str, Any], secret: str = jwt_privatekey) -> str:
         header = {"alg": "RS256"}
         if secret.startswith("-----BEGIN RSA PRIVATE KEY-----"):
-            secret = jwk.dumps(secret, kty="RSA")
+            secret = JsonWebKey.import_key(secret, {"kty": "RSA"})
         result: bytes = jwt.encode(header, payload, secret)
         return result.decode("ascii")