Merge pull request #1858 from openedx/asheehan-edx/fixing-up-intern-work

Fixing up and releasing unfinished intern work

CHANGELOG.rst

@@ -16,6 +16,10 @@ Change Log
 Unreleased
 ----------
 
+[4.1.4]
+-------
+feat: enterprise API Credentials generation endpoints
+
 [4.1.3]
 -------
 fix: bringing changelog and version number back in sync (re-release of 4.1.2).

enterprise/__init__.py

@@ -2,6 +2,6 @@
 Your project description goes here.
 """
 
-__version__ = "4.1.3"
+__version__ = "4.1.4"
 
 default_app_config = "enterprise.apps.EnterpriseConfig"

enterprise/api/utils.py

@@ -3,15 +3,25 @@
 """
 
 from django.conf import settings
+from django.contrib import auth
 from django.utils.translation import gettext as _
 
+from enterprise.constants import (
+    ENTERPRISE_CATALOG_ADMIN_ROLE,
+    ENTERPRISE_DASHBOARD_ADMIN_ROLE,
+    ENTERPRISE_REPORTING_CONFIG_ADMIN_ROLE,
+)
 from enterprise.models import (
+    EnterpriseCustomer,
     EnterpriseCustomerCatalog,
     EnterpriseCustomerInviteKey,
     EnterpriseCustomerReportingConfiguration,
     EnterpriseCustomerUser,
+    EnterpriseFeatureRole,
+    EnterpriseFeatureUserRoleAssignment,
 )
 
+User = auth.get_user_model()
 SERVICE_USERNAMES = (
     'ECOMMERCE_SERVICE_WORKER_USERNAME',
     'ENTERPRISE_SERVICE_WORKER_USERNAME'
@@ -188,3 +198,38 @@ def generate_prompt_for_learner_progress_summary(progress_data):
         prompt = settings.LEARNER_PROGRESS_PROMPT_FOR_NON_ACTIVE_CONTRACT
 
     return prompt.format(**data)
+
+
+def set_application_name_from_user_id(user_id):
+    """
+    Get the enterprise customer user's name given a user id.
+    """
+    try:
+        user = User.objects.get(id=user_id)
+        return f"{user.username}'s Enterprise Credentials"
+    except User.DoesNotExist:
+        return None
+
+
+def has_api_credentials_enabled(enterprise_uuid):
+    """
+    Check whether the enterprise customer can access to api credentials or not
+    """
+    try:
+        return (EnterpriseCustomer
+                .objects.get(uuid=enterprise_uuid)
+                .enable_generation_of_api_credentials)
+    except EnterpriseCustomer.DoesNotExist:
+        return False
+
+
+def assign_feature_roles(user):
+    """
+    Add the ENTERPRISE_DASHBOARD_ADMIN_ROLE, ENTERPRISE_REPORTING_CONFIG_ADMIN_ROLE, ENTERPRISE_CATALOG_ADMIN_ROLE
+    feature roles if the user does not already have them
+    """
+    roles_name = [ENTERPRISE_DASHBOARD_ADMIN_ROLE, ENTERPRISE_REPORTING_CONFIG_ADMIN_ROLE,
+                  ENTERPRISE_CATALOG_ADMIN_ROLE]
+    for role_name in roles_name:
+        feature_role_object, __ = EnterpriseFeatureRole.objects.get_or_create(name=role_name)
+        EnterpriseFeatureUserRoleAssignment.objects.get_or_create(user=user, role=feature_role_object)

enterprise/api/v1/serializers.py

@@ -9,6 +9,7 @@
 
 import pytz
 from edx_rest_api_client.exceptions import HttpClientError
+from oauth2_provider.generators import generate_client_id, generate_client_secret
 from rest_framework import serializers
 from rest_framework.fields import empty
 from rest_framework.settings import api_settings
@@ -1521,3 +1522,46 @@ class LearnerEngagementSerializer(serializers.Serializer):
 
     learner_progress = LearnerProgressSerializer()
     learner_engagement = LearnerEngagementSerializer()
+
+
+class EnterpriseCustomerApiCredentialSerializer(serializers.Serializer):
+    """
+    Serializer for the ``EnterpriseCustomerApiCredential``
+    """
+    class Meta:
+        lookup_field = 'user'
+
+    id = serializers.IntegerField(required=False, read_only=True)
+    name = serializers.CharField(required=False)
+
+    client_id = serializers.CharField(read_only=True, default=generate_client_id())
+    client_secret = serializers.CharField(read_only=True, default=generate_client_secret())
+    authorization_grant_type = serializers.CharField(required=False)
+    client_type = serializers.CharField(required=False)
+    created = serializers.DateTimeField(required=False, read_only=True)
+    updated = serializers.DateTimeField(required=False, read_only=True)
+    redirect_uris = serializers.CharField(required=False)
+    user = UserSerializer(read_only=True)
+
+    def update(self, instance, validated_data):
+        instance.name = validated_data.get('name', instance.name)
+        instance.authorization_grant_type = validated_data.get('authorization_grant_type',
+                                                               instance.authorization_grant_type)
+        instance.client_type = validated_data.get('client_type', instance.client_type)
+        instance.redirect_uris = validated_data.get('redirect_uris', instance.redirect_uris)
+        instance.save()
+        return instance
+
+
+class EnterpriseCustomerApiCredentialRegeneratePatchSerializer(serializers.Serializer):
+    """
+    Serializer for the ``EnterpriseCustomerApiCredential``
+    """
+    class Meta:
+        lookup_field = 'user'
+
+    name = serializers.CharField(required=False)
+    client_id = serializers.CharField(read_only=True, default=generate_client_id())
+    client_secret = serializers.CharField(read_only=True, default=generate_client_secret())
+    redirect_uris = serializers.CharField(required=False)
+    updated = serializers.DateTimeField(required=False, read_only=True)

enterprise/api/v1/urls.py

@@ -12,6 +12,7 @@
     enterprise_catalog_query,
     enterprise_course_enrollment,
     enterprise_customer,
+    enterprise_customer_api_credentials,
     enterprise_customer_branding_configuration,
     enterprise_customer_catalog,
     enterprise_customer_invite_key,
@@ -133,6 +134,20 @@
         analytics_summary.AnalyticsSummaryView.as_view(),
         name='analytics-summary'
     ),
+    re_path(
+        r'^enterprise-customer-api-credentials/(?P<enterprise_uuid>[A-Za-z0-9-]+)/regenerate_credentials$',
+        enterprise_customer_api_credentials.APICredentialsRegenerateViewSet.as_view(
+            {'put': 'update'}
+        ),
+        name='regenerate-api-credentials'
+    ),
+    re_path(
+        r'^enterprise-customer-api-credentials/(?P<enterprise_uuid>[A-Za-z0-9-]+)/$',
+        enterprise_customer_api_credentials.APICredentialsViewSet.as_view(
+            {'get': 'retrieve', 'delete': 'destroy', 'put': 'update', 'post': 'create'}
+        ),
+        name='enterprise-customer-api-credentials'
+    ),
 ]
 
 urlpatterns += router.urls

enterprise/api/v1/views/enterprise_customer_api_credentials.py

@@ -0,0 +1,210 @@
+"""
+Views for the Enterprise Customer API Credentials.
+"""
+from edx_rbac.decorators import permission_required
+from oauth2_provider.generators import generate_client_id, generate_client_secret
+from oauth2_provider.models import get_application_model
+from rest_framework import permissions, status
+from rest_framework.response import Response
+
+from enterprise.api.utils import (
+    assign_feature_roles,
+    get_enterprise_customer_from_user_id,
+    has_api_credentials_enabled,
+    set_application_name_from_user_id,
+)
+from enterprise.api.v1 import serializers
+from enterprise.api.v1.views.base_views import EnterpriseReadWriteModelViewSet
+from enterprise.logging import getEnterpriseLogger
+
+LOGGER = getEnterpriseLogger(__name__)
+
+# Application Model: https://github.com/jazzband/django-oauth-toolkit/blob/master/oauth2_provider/models.py
+Application = get_application_model()
+
+
+class APICredentialsViewSet(EnterpriseReadWriteModelViewSet):
+    """
+    API views for the ``enterprise-customer-api-credentials`` API endpoint.
+    """
+
+    # Verifies the requesting user has the appropriate API permissions
+    permission_classes = (permissions.IsAuthenticated,)
+    # Changes application's pk to be user's pk
+    lookup_field = 'user'
+
+    def get_queryset(self):
+        return Application.objects.filter(user=self.request.user)  # only get current user's record
+
+    def get_serializer_class(self):
+        return serializers.EnterpriseCustomerApiCredentialSerializer
+
+    @permission_required(
+        'enterprise.can_access_admin_dashboard',
+        fn=lambda request, *args, **kwargs: get_enterprise_customer_from_user_id(request.user.id)
+    )
+    def create(self, request, *args, **kwargs):
+        """
+        Creates a new API application credentials and returns the created object.
+
+        Method: POST
+
+        URL: /enterprise/api/v1/enterprise_customer_api_credentials/{enterprise_uuid}
+
+        Returns 201 if a new API application credentials was created.
+        If an application already exists for the user, throw a 409.
+        """
+
+        # Verifies the requesting user is connected to an enterprise that has API credentialing bool set to True
+        user = request.user
+        enterprise_uuid = kwargs['enterprise_uuid']
+        if not enterprise_uuid:
+            return Response({'detail': "Invalid enterprise_uuid"}, status=status.HTTP_400_BAD_REQUEST)
+
+        if not has_api_credentials_enabled(enterprise_uuid):
+            return Response({'detail': 'Can not access API credential viewset.'}, status=status.HTTP_403_FORBIDDEN)
+
+        # Fetches the application for the user
+        # If an application already exists for the user, throw a 409.
+        queryset = self.get_queryset().first()
+        if queryset:
+            return Response({'detail': 'Application exists.'}, status=status.HTTP_409_CONFLICT)
+
+        # Adds the appropriate enterprise related feature roles if they do not already have them
+        assign_feature_roles(user)
+
+        application = Application.objects.create(
+            name=set_application_name_from_user_id(request.user.id),
+            user=request.user,
+            authorization_grant_type="client-credentials",
+            client_type="confidential",
+            client_id=generate_client_id(),
+            client_secret=generate_client_secret()
+        )
+        application.save()
+
+        serializer = self.get_serializer(application)
+        return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+    @permission_required(
+        'enterprise.can_access_admin_dashboard',
+        fn=lambda request, *args, **kwargs: get_enterprise_customer_from_user_id(request.user.id)
+    )
+    def destroy(self, request, *args, **kwargs):
+        """
+        Method: DELETE
+
+        URL: /enterprise/api/v1/enterprise_customer_api_credentials/{enterprise_uuid}
+        """
+        enterprise_uuid = kwargs['enterprise_uuid']
+        if not enterprise_uuid:
+            return Response({'detail': "Invalid enterprise_uuid"}, status=status.HTTP_400_BAD_REQUEST)
+
+        if not has_api_credentials_enabled(enterprise_uuid):
+            return Response({'detail': 'Can not access API credential viewset.'}, status=status.HTTP_403_FORBIDDEN)
+        return super().destroy(self, request, *args, **kwargs)
+
+    @permission_required(
+        'enterprise.can_access_admin_dashboard',
+        fn=lambda request, *args, **kwargs: get_enterprise_customer_from_user_id(request.user.id)
+    )
+    def retrieve(self, request, *args, **kwargs):
+        """
+        Method: GET
+
+        URL: /enterprise/api/v1/enterprise_customer_api_credentials/{enterprise_uuid}
+        """
+        enterprise_uuid = kwargs['enterprise_uuid']
+        if not enterprise_uuid:
+            return Response({'detail': "Invalid enterprise_uuid"}, status=status.HTTP_400_BAD_REQUEST)
+
+        if not has_api_credentials_enabled(enterprise_uuid):
+            return Response({'detail': 'Can not access API credential viewset.'}, status=status.HTTP_403_FORBIDDEN)
+        return super().retrieve(self, request, *args, **kwargs)
+
+    @permission_required(
+        'enterprise.can_access_admin_dashboard',
+        fn=lambda request, *args, **kwargs: get_enterprise_customer_from_user_id(request.user.id)
+    )
+    def update(self, request, *args, **kwargs):
+        """
+        Method: PUT
+
+        URL: /enterprise/api/v1/enterprise_customer_api_credentials/{enterprise_uuid}
+        """
+        # Verifies the requesting user is connected to an enterprise that has API credentialing bool set to True
+        user = request.user
+        enterprise_uuid = kwargs['enterprise_uuid']
+        if not enterprise_uuid:
+            return Response({'detail': "Invalid enterprise_uuid"}, status=status.HTTP_400_BAD_REQUEST)
+
+        if not has_api_credentials_enabled(enterprise_uuid):
+            return Response({'detail': 'Can not access API credential viewset.'}, status=status.HTTP_403_FORBIDDEN)
+
+        queryset = self.get_queryset().first()
+        if not queryset:
+            return Response({'detail': 'Could not find the Application.'}, status=status.HTTP_404_NOT_FOUND)
+
+        instance = Application.objects.get(user=user)
+        expected_fields = {'name', 'client_type', 'redirect_uris', 'authorization_grant_type'}
+        # Throw a 400 if any field to update is not a part of the Application model.
+        for field in request.data.keys():
+            if field not in expected_fields:
+                return Response({'detail': f"Invalid field for update: {field}"}, status=status.HTTP_400_BAD_REQUEST)
+
+        serializer = self.get_serializer(instance, data=request.data)
+        serializer.is_valid(raise_exception=True)
+        serializer.save()
+        return Response(serializer.data, status=status.HTTP_200_OK)
+
+
+class APICredentialsRegenerateViewSet(APICredentialsViewSet):
+    """
+    API views for the ``enterprise-customer-api-credentials`` API endpoint.
+    """
+
+    # Verifies the requesting user has the appropriate API permissions
+    permission_classes = (permissions.IsAuthenticated,)
+    # Changes application's pk to be user's pk
+    lookup_field = 'user'
+
+    def get_queryset(self):
+        return Application.objects.filter(user=self.request.user)  # only get current user's record
+
+    def get_serializer_class(self):
+        return serializers.EnterpriseCustomerApiCredentialRegeneratePatchSerializer
+
+    @permission_required(
+        'enterprise.can_access_admin_dashboard',
+        fn=lambda request, *args, **kwargs: get_enterprise_customer_from_user_id(request.user.id)
+    )
+    def update(self, request, *args, **kwargs):
+        """
+        Method: PUT
+
+        URL: /enterprise/api/v1/enterprise_customer_api_credentials/{enterprise_uuid}/regenerate_credentials
+        """
+        enterprise_uuid = kwargs['enterprise_uuid']
+
+        # Verifies the requesting user is connected to an enterprise that has API credentialing bool set to True
+        if not has_api_credentials_enabled(enterprise_uuid):
+            return Response({'detail': 'Can not access API credential viewset.'}, status=status.HTTP_403_FORBIDDEN)
+
+        # Fetches the application for the user
+        # Throws a 404 if Application record not found
+        application = self.get_queryset().first()
+        if not application:
+            return Response({'detail': 'Could not find the Application.'}, status=status.HTTP_404_NOT_FOUND)
+
+        if 'redirect_uris' not in request.data:
+            return Response({'detail': 'Could not update.'}, status=status.HTTP_400_BAD_REQUEST)
+
+        redirect_uris = request.data.get('redirect_uris')
+        application.redirect_uris = redirect_uris
+        # Calls generate_client_secret and generate_client_id for the user
+        application.client_id = generate_client_id()
+        application.client_secret = generate_client_secret()
+        application.save()
+
+        serializer = self.get_serializer(application)
+        return Response(serializer.data, status=status.HTTP_200_OK)