Add unit tests

changelog.d/16215.bugfix

@@ -0,0 +1 @@
+Fix a bug where admin tokens stopped working with MSC3861 auth delegation was enabled.

synapse/api/auth/msc3861_delegated.py

@@ -28,7 +28,6 @@ from twisted.web.http_headers import Headers
 from synapse.api.auth.base import BaseAuth
 from synapse.api.errors import (
     AuthError,
-    Codes,
     HttpResponseException,
     InvalidClientTokenError,
     OAuthInsufficientScopeError,
@@ -282,17 +281,6 @@ class MSC3861DelegatedAuth(BaseAuth):
                             "Impersonation not possible by a non admin user",
                         )
 
-            # Deny the request if the user account is locked.
-            if not allow_locked and await self.store.get_user_locked_status(
-                requester.user.to_string()
-            ):
-                raise AuthError(
-                    401,
-                    "User account has been locked",
-                    errcode=Codes.USER_LOCKED,
-                    additional_fields={"soft_logout": True},
-                )
-
         if not allow_guest and requester.is_guest:
             raise OAuthInsufficientScopeError([SCOPE_MATRIX_API])
 

tests/handlers/test_oauth_delegation.py

@@ -122,6 +122,7 @@ class MSC3861OAuthDelegation(HomeserverTestCase):
                 "client_id": CLIENT_ID,
                 "client_auth_method": "client_secret_post",
                 "client_secret": CLIENT_SECRET,
+                "admin_token": "admin_token_value",
             }
         }
         return config
@@ -791,3 +792,39 @@ class MSC3861OAuthDelegation(HomeserverTestCase):
         self.expect_unrecognized("GET", "/_synapse/admin/v1/users/foo/admin")
         self.expect_unrecognized("PUT", "/_synapse/admin/v1/users/foo/admin")
         self.expect_unrecognized("POST", "/_synapse/admin/v1/account_validity/validity")
+
+    def test_admin_token(self) -> None:
+        """The handler should return a requester with admin rights when admin_token is used."""
+
+        request = Mock(args={})
+        request.args[b"access_token"] = [b"admin_token_value"]
+        request.requestHeaders.getRawHeaders = mock_getRawHeaders()
+        requester = self.get_success(self.auth.get_user_by_req(request))
+        self.assertEqual(
+            requester.user.to_string(), "@%s:%s" % ("__oidc_admin", SERVER_NAME)
+        )
+        self.assertEqual(requester.is_guest, False)
+        self.assertEqual(requester.device_id, None)
+        self.assertEqual(
+            get_awaitable_result(self.auth.is_server_admin(requester)), True
+        )
+
+    def test_oidc_admin_impersonate_user_id(self) -> None:
+        """The handler should return a requester with the correct user when _oidc_admin_impersonate_user_id param is used."""
+
+        request = Mock(
+            args={
+                b"_oidc_admin_impersonate_user_id": [
+                    ("@foo:" + SERVER_NAME).encode("ascii")
+                ],
+                b"access_token": [b"admin_token_value"],
+            }
+        )
+        request.requestHeaders.getRawHeaders = mock_getRawHeaders()
+        requester = self.get_success(self.auth.get_user_by_req(request))
+        self.assertEqual(requester.user.to_string(), "@%s:%s" % ("foo", SERVER_NAME))
+        self.assertEqual(requester.is_guest, False)
+        self.assertEqual(requester.device_id, None)
+        self.assertEqual(
+            get_awaitable_result(self.auth.is_server_admin(requester)), False
+        )