OpenConceptLab/ocl_issues#1338 | API to migrate user from django to SSO

core/common/services.py

@@ -308,7 +308,7 @@ def credential_representation_from_hash(hash_, temporary=False):
         }
 
     @classmethod
-    def add_user(cls, user):
+    def add_user(cls, user, username=None, password=None):
         response = requests.post(
             cls.USERS_URL,
             json=dict(
@@ -321,7 +321,7 @@ def add_user(cls, user):
                 credentials=[cls.credential_representation_from_hash(hash_=user.password)]
             ),
             verify=False,
-            headers=OIDCAuthService.get_admin_headers()
+            headers=OIDCAuthService.get_admin_headers(username=username, password=password)
         )
         if response.status_code == 201:
             return True
@@ -335,13 +335,13 @@ def get_token(self):
         return self.token_type + ' ' + get(token, 'access_token')
 
     @staticmethod
-    def get_admin_token():
+    def get_admin_token(username=None, password=None):
         response = requests.post(
             OIDCAuthService.OIDP_ADMIN_TOKEN_URL,
             data=dict(
                 grant_type='password',
-                username=settings.KEYCLOAK_ADMIN,
-                password=settings.KEYCLOAK_ADMIN_PASSWORD,
+                username=username or settings.KEYCLOAK_ADMIN,
+                password=password or settings.KEYCLOAK_ADMIN_PASSWORD,
                 client_id='admin-cli'
             ),
             verify=False,
@@ -385,8 +385,8 @@ def exchange_code_for_token(code, redirect_uri):
         return response.json()
 
     @staticmethod
-    def get_admin_headers():
-        return dict(Authorization=f'Bearer {OIDCAuthService.get_admin_token()}')
+    def get_admin_headers(**kwargs):
+        return dict(Authorization=f'Bearer {OIDCAuthService.get_admin_token(**kwargs)}')
 
     def get_user_headers(self):
         return dict(Authorization=self.get_token())

core/users/urls.py

@@ -15,6 +15,11 @@
         views.UserDetailView.as_view(),
         name='userprofile-detail'
     ),
+    path(
+        '<str:user>/sso-migrate/',
+        views.SSOMigrateView.as_view(),
+        name='userprofile-sso-migrate'
+    ),
     path(
         '<str:user>/verify/<str:verification_token>/',
         views.UserEmailVerificationView.as_view(),

core/users/views.py

@@ -62,6 +62,29 @@ def post(request):
             OIDCAuthService.exchange_code_for_token(code, redirect_uri))
 
 
+class SSOMigrateView(APIView):
+    permission_classes = (AllowAny, )
+
+    def get_object(self):
+        username = self.kwargs.get('user')
+        user = UserProfile.objects.filter(username=username).first()
+        if not user:
+            raise Http404()
+        return user
+
+    def post(self, request, **kwargs):  # pylint: disable=unused-argument
+        username = request.data.get('username')
+        password = request.data.get('password')
+        if not username or not password:
+            return Response(
+                dict(error='keycloak admin username/password are required'),
+                status=status.HTTP_400_BAD_REQUEST
+            )
+        user = self.get_object()
+        result = OIDCAuthService.add_user(user=user, username=username, password=password)
+        return Response(result)
+
+
 class TokenAuthenticationView(ObtainAuthToken):
     """Implementation of ObtainAuthToken with last_login update"""