OpenConceptLab/ocl_issues#1338 | removed unused code

core/common/authentication.py

@@ -1,10 +1,12 @@
+from django.conf import settings
+from pydash import get
 from rest_framework.authentication import BaseAuthentication, TokenAuthentication
 
 
 class OCLAuthentication(BaseAuthentication):
     def get_auth_class(self, request):
         from core.common.services import AuthService
-        if AuthService.is_valid_django_token(request):
+        if AuthService.is_valid_django_token(request) or get(settings, 'TEST_MODE', False):
             klass = TokenAuthentication
         else:
             klass = AuthService.get().authentication_class

core/common/backends.py

@@ -1,3 +1,4 @@
+from django.conf import settings
 from django.contrib.auth.backends import ModelBackend
 from mozilla_django_oidc.auth import OIDCAuthenticationBackend
 from pydash import get
@@ -53,7 +54,7 @@ def get_auth_backend(self, request=None):
             return get(self, '_authentication_backend')
 
         from core.common.services import AuthService
-        if AuthService.is_valid_django_token(request):
+        if AuthService.is_valid_django_token(request) or get(settings, 'TEST_MODE', False):
             klass = ModelBackend
         else:
             klass = AuthService.get().authentication_backend_class

core/common/services.py

@@ -10,7 +10,6 @@
 from django.contrib.auth.backends import ModelBackend
 from django.core.files.base import ContentFile
 from django.db import connection
-from django.http import Http404
 from mozilla_django_oidc.contrib.drf import OIDCAuthentication
 from pydash import get
 from rest_framework.authentication import TokenAuthentication
@@ -317,7 +316,7 @@ def credential_representation_from_hash(hash_, temporary=False):
         }
 
     @classmethod
-    def add_user(cls, user, username=None, password=None):
+    def add_user(cls, user, username, password):
         response = requests.post(
             cls.USERS_URL,
             json=dict(
@@ -338,46 +337,22 @@ def add_user(cls, user, username=None, password=None):
         return response.json()
 
     def get_token(self):
-        token = self.user.get_oidc_token(self.password)
-        if token is False:
-            return token
-        return self.token_type + ' ' + get(token, 'access_token')
+        return None
 
     @staticmethod
-    def get_admin_token(username=None, password=None):
+    def get_admin_token(username, password):
         response = requests.post(
             OIDCAuthService.OIDP_ADMIN_TOKEN_URL,
             data=dict(
                 grant_type='password',
-                username=username or settings.KEYCLOAK_ADMIN,
-                password=password or settings.KEYCLOAK_ADMIN_PASSWORD,
+                username=username,
+                password=password,
                 client_id='admin-cli'
             ),
             verify=False,
         )
         return response.json().get('access_token')
 
-    def logout(self, token):
-        user_info = self.get_user_info(token)
-        user_id = get(user_info, 'sub')
-        if user_id:
-            requests.post(
-                f"{self.USERS_URL}/{user_id}/logout",
-                json=dict(user=user_id, realm=settings.OIDC_REALM),
-                headers=self.get_admin_headers()
-            )
-
-        return user_info
-
-    @staticmethod
-    def get_user_info(token):
-        response = requests.post(
-            settings.OIDC_OP_USER_ENDPOINT,
-            verify=False,
-            headers=dict(Authorization=token)
-        )
-        return response.json()
-
     @staticmethod
     def exchange_code_for_token(code, redirect_uri):
         response = requests.post(
@@ -397,90 +372,17 @@ def exchange_code_for_token(code, redirect_uri):
     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())
-
     @staticmethod
-    def create_user(data):
-        response = requests.post(
-            OIDCAuthService.USERS_URL,
-            json=dict(
-                enabled=True,
-                emailVerified=False,
-                firstName=data.get('first_name'),
-                lastName=data.get('last_name'),
-                email=data.get('email'),
-                username=data.get('username'),
-                credentials=[dict(type='password', value=data.get('password'), temporary=False)]
-            ),
-            verify=False,
-            headers=OIDCAuthService.get_admin_headers()
-        )
-        if response.status_code == 201:
-            return True
-
-        return response.json()
-
-    def __get_all_users(self, headers=None):
-        response = requests.get(
-            self.USERS_URL,
-            verify=False,
-            headers=headers or self.get_admin_headers()
-        )
-        return response.json()
-
-    def __get_user_info(self, headers=None):
-        users = self.__get_all_users(headers)
-        return next((user for user in users if user['username'] == self.username), None)
-
-    def __get_user_id(self, headers=None):
-        user_info = self.__get_user_info(headers)
-        return get(user_info, 'id')
-
-    def mark_verified(self, **kwargs):
-        response = self.update_user(dict(emailVerified=True))
-        if response.status_code < 300:
-            return super().mark_verified(**kwargs)
-        return response.json()
-
-    def update_user(self, data):
-        admin_headers = self.get_admin_headers()
-        oid_user_id = self.__get_user_id(admin_headers)
-        if not oid_user_id:
-            raise Http404()
-        response = requests.put(
-            f"{self.USERS_URL}/{oid_user_id}",
-            json=data,
-            verify=False,
-            headers=admin_headers
-        )
-        return response
-
-    def update_password(self, password):
-        admin_headers = self.get_admin_headers()
-        oid_user_id = self.__get_user_id(admin_headers)
-        if not oid_user_id:
-            raise Http404()
-
-        requests.put(
-            f"{self.USERS_URL}/{oid_user_id}/disable-credential-types",
-            json=['password'],
-            verify=False,
-            headers=admin_headers
-        )
-
-        response = requests.put(
-            f"{self.USERS_URL}/{oid_user_id}/reset-password",
-            json=dict(type='password', temporary=False, value=password),
-            verify=False,
-            headers=admin_headers
-        )
-        if response.status_code < 300:
-            return super().update_password(password)
-        return response.json()
+    def create_user(_):
+        """In OID auth, user signup needs to happen in OID first"""
+        pass  # pylint: disable=unnecessary-pass
 
 
 class AuthService:
+    @staticmethod
+    def is_sso_enabled():
+        return get(settings, 'OIDC_SERVER_URL') and not get(settings, 'TEST_MODE', False)
+
     @staticmethod
     def get(**kwargs):
         if get(settings, 'OIDC_SERVER_URL'):

core/settings.py

@@ -416,6 +416,4 @@
 OIDC_CREATE_USER = True
 LOGIN_REDIRECT_URL = os.environ.get('LOGIN_REDIRECT_URL', 'http://localhost:4000')
 LOGOUT_REDIRECT_URL = os.environ.get('LOGOUT_REDIRECT_URL', 'http://localhost:4000')
-KEYCLOAK_ADMIN = os.environ.get('KEYCLOAK_ADMIN', 'root')
-KEYCLOAK_ADMIN_PASSWORD = os.environ.get('KEYCLOAK_ADMIN_PASSWORD', 'Root123')
 OIDC_CALLBACK_CLASS = 'core.users.views.OCLOIDCAuthenticationCallbackView'

core/users/models.py

@@ -1,8 +1,6 @@
 import uuid
 from datetime import datetime
 
-import requests
-from django.conf import settings
 from django.contrib.auth.models import AbstractUser
 from django.contrib.auth.password_validation import validate_password
 from django.core.exceptions import ValidationError
@@ -112,22 +110,6 @@ def get_token(self):
         token = Token.objects.filter(user_id=self.id).first() or self.__create_token()
         return token.key
 
-    def get_oidc_token(self, password):
-        response = requests.post(
-            settings.OIDC_OP_TOKEN_ENDPOINT,
-            data=dict(
-                username=self.username,
-                password=password,
-                grant_type='password',
-                client_id=settings.OIDC_RP_CLIENT_ID,
-                client_secret=settings.OIDC_RP_CLIENT_SECRET
-            ),
-            verify=False,
-        )
-        if response.status_code != 200:
-            return False
-        return response.json()
-
     def set_token(self, token):
         self.__delete_token()
         Token.objects.create(user=self, key=token)

core/users/views.py

@@ -1,5 +1,6 @@
 import uuid
 
+from django.conf import settings
 from django.contrib.auth.models import update_last_login
 from django.contrib.auth.password_validation import validate_password
 from django.core.exceptions import ValidationError
@@ -62,7 +63,8 @@ def post(request):
             OIDCAuthService.exchange_code_for_token(code, redirect_uri))
 
 
-class SSOMigrateView(APIView):
+# This API is only to migrate users from Django to OID, requires OID admin credentials in payload
+class SSOMigrateView(APIView):  # pragma: no cover
     permission_classes = (AllowAny, )
 
     def get_object(self):
@@ -98,37 +100,40 @@ class TokenAuthenticationView(ObtainAuthToken):
 
     @swagger_auto_schema(request_body=AuthTokenSerializer)
     def post(self, request, *args, **kwargs):
-        username = request.data.get('username')
-        password = request.data.get('password')
+        if AuthService.is_sso_enabled():
+            raise Http400(
+                dict(error=["Single Sign On is enabled in this environment. Cannot login via API directly."]))
 
-        if not username or not password:
-            raise Http400(dict(non_field_errors=['Must include "username" and "password".']))
+        user = UserProfile.objects.filter(username=request.data.get('username')).first()
 
-        auth_service = AuthService.get(username=username, password=password)
-        user = auth_service.user
+        if not user or not user.check_password(request.data.get('password')):
+            raise Http400(dict(non_field_errors=["Unable to log in with provided credentials."]))
 
         if not user.is_active:
             user.verify()
             return Response(
-                {'detail': REACTIVATE_USER_MESSAGE, 'email': user.email}, status=status.HTTP_401_UNAUTHORIZED
+                {
+                    'detail': REACTIVATE_USER_MESSAGE,
+                    'email': user.email
+                }, status=status.HTTP_401_UNAUTHORIZED
             )
         if not user.verified:
             user.send_verification_email()
             return Response(
-                {'detail': VERIFY_EMAIL_MESSAGE, 'email': user.email}, status=status.HTTP_401_UNAUTHORIZED
+                {
+                    'detail': VERIFY_EMAIL_MESSAGE,
+                    'email': user.email
+                }, status=status.HTTP_401_UNAUTHORIZED
             )
 
-        token = auth_service.get_token()
-
-        if token is False:
-            raise Http400(dict(non_field_errors=["Unable to log in with provided credentials."]))
+        result = super().post(request, *args, **kwargs)
 
         try:
             update_last_login(None, user)
         except:  # pylint: disable=bare-except
             pass
 
-        return Response(dict(token=token))
+        return result
 
 
 class UserBaseView(BaseAPIView):
@@ -244,6 +249,9 @@ def post(self, request, *args, **kwargs):  # pylint: disable=unused-argument
         return Response({'detail': VERIFY_EMAIL_MESSAGE}, status=status.HTTP_201_CREATED, headers=headers)
 
     def perform_create(self, serializer):
+        if AuthService.is_sso_enabled():
+            raise Http400(
+                dict(error=["Single Sign On is enabled in this environment. Cannot signup via API directly."]))
         data = self.request.data
         try:
             validate_password(data.get('password'))
@@ -252,9 +260,9 @@ def perform_create(self, serializer):
             return
 
         response = AuthService.get().create_user(data)
-        if response is True:
+        if response is True or get(settings, 'TEST_MODE', False):
             super().perform_create(serializer)
-        else:
+        elif response:
             serializer._errors = response  # pylint: disable=protected-access
 
 
@@ -297,6 +305,10 @@ def post(self, request, *args, **kwargs):  # pylint: disable=unused-argument
     def put(self, request, *args, **kwargs):  # pylint: disable=unused-argument
         """Resets password"""
 
+        if AuthService.is_sso_enabled():
+            raise Http400(
+                dict(error=["Single Sign On is enabled in this environment. Cannot reset password via API directly."]))
+
         token = request.data.get('token', None)
         password = request.data.get('new_password', None)
         if not token or not password:
@@ -312,8 +324,7 @@ def put(self, request, *args, **kwargs):  # pylint: disable=unused-argument
         except ValidationError as ex:
             return Response(dict(errors=ex.messages), status=status.HTTP_400_BAD_REQUEST)
 
-        service = AuthService.get(user=user)
-        result = service.update_password(password)
+        result = AuthService.get(user=user).update_password(password)
         if get(result, 'errors'):
             return Response(result, status=status.HTTP_400_BAD_REQUEST)
 

docker-compose.yml

@@ -52,8 +52,6 @@ services:
       - OIDC_SERVER_URL=${OIDC_SERVER_URL-http://localhost:8080}
       - OIDC_SERVER_INTERNAL_URL=${OIDC_SERVER_INTERNAL_URL-http://host.docker.internal:8080} # only for dev env
       - OIDC_REALM=${OIDC_REALM-ocl}
-      - KEYCLOAK_ADMIN=${KEYCLOAK_ADMIN-root}
-      - KEYCLOAK_ADMIN_PASSWORD=${KEYCLOAK_ADMIN_PASSWORD-Root123}
     healthcheck:
       test: "curl --silent --fail http://localhost:8000/version/ || exit 1"
   celery: