MSC3861: allow impersonation by an admin using a query param (#16132)

changelog.d/16132.misc

@@ -0,0 +1 @@
+MSC3861: allow impersonation by an admin user using `_oidc_admin_impersonate_user_id` query parameter.

synapse/api/auth/msc3861_delegated.py

@@ -246,7 +246,7 @@ class MSC3861DelegatedAuth(BaseAuth):
         return introspection_token
 
     async def is_server_admin(self, requester: Requester) -> bool:
-        return "urn:synapse:admin:*" in requester.scope
+        return SCOPE_SYNAPSE_ADMIN in requester.scope
 
     async def get_user_by_req(
         self,
@@ -263,6 +263,25 @@ class MSC3861DelegatedAuth(BaseAuth):
             # so that we don't provision the user if they don't have enough permission:
             requester = await self.get_user_by_access_token(access_token, allow_expired)
 
+            # Allow impersonation by an admin user using `_oidc_admin_impersonate_user_id` query parameter
+            if request.args is not None:
+                user_id_params = request.args.get(b"_oidc_admin_impersonate_user_id")
+                if user_id_params:
+                    if await self.is_server_admin(requester):
+                        user_id_str = user_id_params[0].decode("ascii")
+                        impersonated_user_id = UserID.from_string(user_id_str)
+                        logging.info(f"Admin impersonation of user {user_id_str}")
+                        requester = create_requester(
+                            user_id=impersonated_user_id,
+                            scope=[SCOPE_MATRIX_API],
+                            authenticated_entity=requester.user.to_string(),
+                        )
+                    else:
+                        raise AuthError(
+                            401,
+                            "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()
@@ -290,14 +309,14 @@ class MSC3861DelegatedAuth(BaseAuth):
             # XXX: This is a temporary solution so that the admin API can be called by
             # the OIDC provider. This will be removed once we have OIDC client
             # credentials grant support in matrix-authentication-service.
-            logging.info("Admin toked used")
+            logging.info("Admin token used")
             # XXX: that user doesn't exist and won't be provisioned.
             # This is mostly fine for admin calls, but we should also think about doing
             # requesters without a user_id.
             admin_user = UserID("__oidc_admin", self._hostname)
             return create_requester(
                 user_id=admin_user,
-                scope=["urn:synapse:admin:*"],
+                scope=[SCOPE_SYNAPSE_ADMIN],
             )
 
         try:

tests/handlers/test_oauth_delegation.py

@@ -340,6 +340,41 @@ class MSC3861OAuthDelegation(HomeserverTestCase):
             get_awaitable_result(self.auth.is_server_admin(requester)), False
         )
 
+    def test_active_user_admin_impersonation(self) -> None:
+        """The handler should return a requester with normal user rights
+        and an user ID matching the one specified in query param `user_id`"""
+
+        self.http_client.request = simple_async_mock(
+            return_value=FakeResponse.json(
+                code=200,
+                payload={
+                    "active": True,
+                    "sub": SUBJECT,
+                    "scope": " ".join([SYNAPSE_ADMIN_SCOPE, MATRIX_USER_SCOPE]),
+                    "username": USERNAME,
+                },
+            )
+        )
+        request = Mock(args={})
+        request.args[b"access_token"] = [b"mockAccessToken"]
+        impersonated_user_id = f"@{USERNAME}:{SERVER_NAME}"
+        request.args[b"_oidc_admin_impersonate_user_id"] = [
+            impersonated_user_id.encode("ascii")
+        ]
+        request.requestHeaders.getRawHeaders = mock_getRawHeaders()
+        requester = self.get_success(self.auth.get_user_by_req(request))
+        self.http_client.get_json.assert_called_once_with(WELL_KNOWN)
+        self.http_client.request.assert_called_once_with(
+            method="POST", uri=INTROSPECTION_ENDPOINT, data=ANY, headers=ANY
+        )
+        self._assertParams()
+        self.assertEqual(requester.user.to_string(), impersonated_user_id)
+        self.assertEqual(requester.is_guest, False)
+        self.assertEqual(requester.device_id, None)
+        self.assertEqual(
+            get_awaitable_result(self.auth.is_server_admin(requester)), False
+        )
+
     def test_active_user_with_device(self) -> None:
         """The handler should return a requester with normal user rights and a device ID."""