diff options
| -rw-r--r-- | changelog.d/16132.misc | 1 | ||||
| -rw-r--r-- | synapse/api/auth/msc3861_delegated.py | 25 | ||||
| -rw-r--r-- | tests/handlers/test_oauth_delegation.py | 35 |
3 files changed, 58 insertions, 3 deletions
diff --git a/changelog.d/16132.misc b/changelog.d/16132.misc new file mode 100644 index 0000000000..aca26079d8 --- /dev/null +++ b/changelog.d/16132.misc @@ -0,0 +1 @@ +MSC3861: allow impersonation by an admin user using `_oidc_admin_impersonate_user_id` query parameter. diff --git a/synapse/api/auth/msc3861_delegated.py b/synapse/api/auth/msc3861_delegated.py index 18875f2c81..4bdfe31b22 100644 --- a/synapse/api/auth/msc3861_delegated.py +++ b/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: diff --git a/tests/handlers/test_oauth_delegation.py b/tests/handlers/test_oauth_delegation.py index 82c26e303f..1456b675a7 100644 --- a/tests/handlers/test_oauth_delegation.py +++ b/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.""" |