Merge pull request #2216 from openedx/sameeramin/ENT-8971

feat: populate encrypted client id and secret

CHANGELOG.rst

@@ -17,6 +17,10 @@ Unreleased
 ----------
 * nothing unreleased
 
+[4.23.14]
+----------
+* feat: populate encrypted client id and secret for canvas integration and remove references to unencrypted fields
+
 [4.23.13]
 ----------
 * feat: added encrypted columns for user credentials for SAP config

enterprise/__init__.py

@@ -2,4 +2,4 @@
 Your project description goes here.
 """
 
-__version__ = "4.23.13"
+__version__ = "4.23.14"

enterprise/signals.py

@@ -455,17 +455,6 @@ def generate_default_orchestration_record_display_name(sender, instance, **kwarg
             instance.display_name = f'SSO-config-{instance.identity_provider}-{num_records_for_customer + 1}'
 
 
-@receiver(pre_save, sender=CanvasEnterpriseCustomerConfiguration)
-def mirror_id_and_secret_to_decrypted_fields(sender, instance, **kwargs):   # pylint: disable=unused-argument
-    """
-    Mirror the client_id and client_secret to the decrypted fields.
-    """
-    if instance.client_id != instance.decrypted_client_id:
-        instance.decrypted_client_id = instance.client_id
-    if instance.client_secret != instance.decrypted_client_secret:
-        instance.decrypted_client_secret = instance.client_secret
-
-
 # Don't connect this receiver if we dont have access to CourseEnrollment model
 if CourseEnrollment is not None:
     post_save.connect(create_enterprise_enrollment_receiver, sender=CourseEnrollment)

integrated_channels/api/v1/canvas/serializers.py

@@ -1,6 +1,8 @@
 """
 Serializers for Canvas.
 """
+from rest_framework import serializers
+
 from integrated_channels.api.serializers import EnterpriseCustomerPluginConfigSerializer
 from integrated_channels.canvas.models import CanvasEnterpriseCustomerConfiguration
 
@@ -9,12 +11,15 @@ class CanvasEnterpriseCustomerConfigurationSerializer(EnterpriseCustomerPluginCo
     class Meta:
         model = CanvasEnterpriseCustomerConfiguration
         extra_fields = (
-            'client_id',
-            'client_secret',
+            'encrypted_client_id',
+            'encrypted_client_secret',
             'canvas_account_id',
             'canvas_base_url',
             'refresh_token',
             'uuid',
             'oauth_authorization_url',
         )
         fields = EnterpriseCustomerPluginConfigSerializer.Meta.fields + extra_fields
+
+    encrypted_client_id = serializers.CharField(required=False, allow_blank=False, read_only=False)
+    encrypted_client_secret = serializers.CharField(required=False, allow_blank=False, read_only=False)

integrated_channels/canvas/client.py

@@ -959,8 +959,8 @@ def _get_oauth_access_token(self):
             HTTPError: If we received a failure response code from Canvas.
             ClientError: If an unexpected response format was received that we could not parse.
         """
-        client_id = self.enterprise_configuration.client_id
-        client_secret = self.enterprise_configuration.client_secret
+        client_id = self.enterprise_configuration.decrypted_client_id
+        client_secret = self.enterprise_configuration.decrypted_client_secret
 
         if not client_id:
             raise ClientError(

integrated_channels/canvas/migrations/0037_canvasenterprisecustomerconfiguration_copy_id_and_secret_and_more.py

@@ -0,0 +1,26 @@
+# Generated by Django 4.2.15 on 2024-08-26 13:57
+
+from django.db import migrations
+
+
+def populate_decrypted_client_id_and_secret(apps, schema_editor):  # pragma: no cover
+    """
+    Populate the decrypted_client_id and decrypted_client_secret fields for all
+    existing CanvasEnterpriseCustomerConfiguration
+    """
+    CanvasEnterpriseCustomerConfiguration = apps.get_model('canvas', 'CanvasEnterpriseCustomerConfiguration')
+    for config in CanvasEnterpriseCustomerConfiguration.objects.all():
+        config.decrypted_client_id = config.client_id
+        config.decrypted_client_secret = config.client_secret
+        config.save()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('canvas', '0036_canvasenterprisecustomerconfiguration_decrypted_client_id_and_more'),
+    ]
+
+    operations = [
+        migrations.RunPython(populate_decrypted_client_id_and_secret, reverse_code=migrations.RunPython.noop),
+    ]

integrated_channels/canvas/models.py

@@ -11,6 +11,7 @@
 
 from django.conf import settings
 from django.db import models
+from django.utils.encoding import force_bytes, force_str
 from django.utils.translation import gettext_lazy as _
 
 from integrated_channels.canvas.exporters.content_metadata import CanvasContentMetadataExporter
@@ -57,6 +58,28 @@ class CanvasEnterpriseCustomerConfiguration(EnterpriseCustomerPluginConfiguratio
         null=True,
     )
 
+    @property
+    def encrypted_client_id(self):
+        """
+        Returns encrypted client_id as a string.
+        The data is encrypted in the DB at rest, but unencrypted in the app when retrieved through
+        `decrypted_client_id`. This method will encrypt the `client_id` again before sending.
+        """
+        if self.decrypted_client_id:
+            return force_str(
+                self._meta.get_field('decrypted_client_id').fernet.encrypt(
+                    force_bytes(self.decrypted_client_id)
+                )
+            )
+        return self.decrypted_client_id
+
+    @encrypted_client_id.setter
+    def encrypted_client_id(self, value):
+        """
+        Set the `decrypted_client_id` from the encrypted value.
+        """
+        self.decrypted_client_id = value
+
     client_secret = models.CharField(
         max_length=255,
         blank=True,
@@ -80,6 +103,28 @@ class CanvasEnterpriseCustomerConfiguration(EnterpriseCustomerPluginConfiguratio
         null=True,
     )
 
+    @property
+    def encrypted_client_secret(self):
+        """
+        Returns encrypted client_secret as a string.
+        The data is encrypted in the DB at rest, but unencrypted in the app when retrieved through
+        `decrypted_client_secret`. This method will encrypt the `client_secret` again before sending.
+        """
+        if self.decrypted_client_secret:
+            return force_str(
+                self._meta.get_field('decrypted_client_secret').fernet.encrypt(
+                    force_bytes(self.decrypted_client_secret)
+                )
+            )
+        return self.decrypted_client_secret
+
+    @encrypted_client_secret.setter
+    def encrypted_client_secret(self, value):
+        """
+        Set the `decrypted_client_secret` from the encrypted value.
+        """
+        self.decrypted_client_secret = value
+
     canvas_account_id = models.BigIntegerField(
         null=True,
         blank=True,
@@ -132,11 +177,11 @@ def oauth_authorization_url(self):
             obj: The instance of CanvasEnterpriseCustomerConfiguration
                 being rendered with this admin form.
         """
-        if self.canvas_base_url and self.client_id:
+        if self.canvas_base_url and self.decrypted_client_id:
             return (f'{self.canvas_base_url}/login/oauth2/auth'
                     f'?redirect_uri={LMS_OAUTH_REDIRECT_URL}&'
                     f'response_type=code&'
-                    f'client_id={self.client_id}&state={self.uuid}')
+                    f'client_id={self.decrypted_client_id}&state={self.uuid}')
         else:
             return None
 
@@ -151,9 +196,9 @@ def is_valid(self):
         """
         missing_items = {'missing': []}
         incorrect_items = {'incorrect': []}
-        if not self.client_id:
+        if not self.decrypted_client_id:
             missing_items.get('missing').append('client_id')
-        if not self.client_secret:
+        if not self.decrypted_client_secret:
             missing_items.get('missing').append('client_secret')
         if not self.canvas_base_url:
             missing_items.get('missing').append('canvas_base_url')

integrated_channels/canvas/views.py

@@ -113,8 +113,8 @@ def get(self, request, *args, **kwargs):
 
         access_token_request_params = {
             'grant_type': 'authorization_code',
-            'client_id': enterprise_config.client_id,
-            'client_secret': enterprise_config.client_secret,
+            'client_id': enterprise_config.decrypted_client_id,
+            'client_secret': enterprise_config.decrypted_client_secret,
             'redirect_uri': settings.LMS_INTERNAL_ROOT_URL + "/canvas/oauth-complete",
             'code': client_code,
         }

tests/test_integrated_channels/test_api/test_canvas/test_views.py

@@ -33,8 +33,8 @@ def setUp(self):
 
         self.enterprise_customer_conf = factories.CanvasEnterpriseCustomerConfigurationFactory(
             enterprise_customer=self.enterprise_customer,
-            client_id='ayy',
-            client_secret='lmao',
+            encrypted_client_id='ayy',
+            encrypted_client_secret='lmao',
         )
 
     @mock.patch('enterprise.rules.crum.get_current_request')
@@ -120,6 +120,8 @@ def test_update(self, mock_current_request):
         payload = {
             'canvas_account_id': 1000,
             'enterprise_customer': ENTERPRISE_ID,
+            'encrypted_client_id': '',
+            'encrypted_client_secret': '',
         }
         response = self.client.put(url, payload)
         self.enterprise_customer_conf.refresh_from_db()
@@ -169,8 +171,8 @@ def test_is_valid_field(self, mock_current_request):
         assert missing.get('missing') == ['refresh_token']
         assert incorrect.get('incorrect') == ['canvas_base_url', 'display_name']
 
-        self.enterprise_customer_conf.client_id = ''
-        self.enterprise_customer_conf.client_secret = ''
+        self.enterprise_customer_conf.decrypted_client_id = ''
+        self.enterprise_customer_conf.decrypted_client_secret = ''
         self.enterprise_customer_conf.canvas_base_url = ''
         self.enterprise_customer_conf.canvas_account_id = None
         self.enterprise_customer_conf.save()
@@ -184,8 +186,8 @@ def test_is_valid_field(self, mock_current_request):
         # Add a refresh token and assert that is_valid now passes
         self.enterprise_customer_conf.refresh_token = 'ayylmao'
         self.enterprise_customer_conf.canvas_base_url = 'http://lovely.com'
-        self.enterprise_customer_conf.client_id = '1'
-        self.enterprise_customer_conf.client_secret = '1'
+        self.enterprise_customer_conf.decrypted_client_id = '1'
+        self.enterprise_customer_conf.decrypted_client_secret = '1'
         self.enterprise_customer_conf.canvas_account_id = 1
         self.enterprise_customer_conf.display_name = 'nice<3'
         self.enterprise_customer_conf.save()

tests/test_integrated_channels/test_canvas/test_client.py

@@ -96,13 +96,13 @@ def setUp(self):
 
         self.course_api_path = "/api/v1/provider/content/course"
         self.course_url = urljoin(self.url_base, self.course_api_path)
-        self.client_id = "client_id"
-        self.client_secret = "client_secret"
+        self.decrypted_client_id = "client_id"
+        self.decrypted_client_secret = "client_secret"
         self.access_token = "access_token"
         self.refresh_token = "refresh_token"
         self.enterprise_config = factories.CanvasEnterpriseCustomerConfigurationFactory(
-            client_id=self.client_id,
-            client_secret=self.client_secret,
+            decrypted_client_id=self.decrypted_client_id,
+            decrypted_client_secret=self.decrypted_client_secret,
             canvas_account_id=self.account_id,
             canvas_base_url=self.url_base,
             refresh_token=self.refresh_token,
@@ -426,14 +426,14 @@ def test_create_client_session_with_oauth_access_key(self):
 
     def test_client_instantiation_fails_without_client_id(self):
         with pytest.raises(ClientError) as client_error:
-            self.enterprise_config.client_id = None
+            self.enterprise_config.decrypted_client_id = None
             canvas_api_client = CanvasAPIClient(self.enterprise_config)
             canvas_api_client._create_session()  # pylint: disable=protected-access
         assert client_error.value.message == "Failed to generate oauth access token: Client ID required."
 
     def test_client_instantiation_fails_without_client_secret(self):
         with pytest.raises(ClientError) as client_error:
-            self.enterprise_config.client_secret = None
+            self.enterprise_config.decrypted_client_secret = None
             canvas_api_client = CanvasAPIClient(self.enterprise_config)
             canvas_api_client._create_session()  # pylint: disable=protected-access
         assert client_error.value.message == "Failed to generate oauth access token: Client secret required."
@@ -1083,8 +1083,8 @@ def test_health_check_invalid_config(self):
         Test the client health check with invalid config
         """
         bad_enterprise_config = factories.CanvasEnterpriseCustomerConfigurationFactory(
-            client_id=self.client_id,
-            client_secret=self.client_secret,
+            decrypted_client_id=self.decrypted_client_id,
+            decrypted_client_secret=self.decrypted_client_secret,
             canvas_account_id=self.account_id,
             canvas_base_url='Not a valid url',
             refresh_token=self.refresh_token,

tests/test_integrated_channels/test_canvas/test_utils.py

@@ -28,13 +28,13 @@ def setUp(self):
         self.account_id = random.randint(9223372036854775800, 9223372036854775807)
         self.course_id = "edx+111"
         self.url_base = "http://betatest.instructure.com"
-        self.client_id = "client_id"
-        self.client_secret = "client_secret"
+        self.decrypted_client_id = "client_id"
+        self.decrypted_client_secret = "client_secret"
         self.access_token = "access_token"
         self.refresh_token = "refresh_token"
         self.enterprise_config = factories.CanvasEnterpriseCustomerConfigurationFactory(
-            client_id=self.client_id,
-            client_secret=self.client_secret,
+            decrypted_client_id=self.decrypted_client_id,
+            decrypted_client_secret=self.decrypted_client_secret,
             canvas_account_id=self.account_id,
             canvas_base_url=self.url_base,
             refresh_token=self.refresh_token,

tests/test_integrated_channels/test_canvas/test_views.py

@@ -56,27 +56,47 @@ def setUp(self):
         self.refresh_token = 'test-refresh-token'
         self.urlbase = reverse('canvas-oauth-complete')
 
-        CanvasEnterpriseCustomerConfiguration.objects.get_or_create(
-            uuid=SINGLE_CANVAS_CONFIG['uuid'],
-            client_id=SINGLE_CANVAS_CONFIG['client_id'],
-            client_secret=SINGLE_CANVAS_CONFIG['client_secret'],
-            canvas_account_id=SINGLE_CANVAS_CONFIG['canvas_account_id'],
-            canvas_base_url=SINGLE_CANVAS_CONFIG['canvas_base_url'],
-            enterprise_customer=self.enterprise_customer,
-            active=True,
-            enterprise_customer_id=ENTERPRISE_ID,
-        )
+        try:
+            CanvasEnterpriseCustomerConfiguration.objects.get(
+                uuid=SINGLE_CANVAS_CONFIG['uuid'],
+                canvas_account_id=SINGLE_CANVAS_CONFIG['canvas_account_id'],
+                canvas_base_url=SINGLE_CANVAS_CONFIG['canvas_base_url'],
+                enterprise_customer=self.enterprise_customer,
+                active=True,
+                enterprise_customer_id=ENTERPRISE_ID,
+            )
+        except CanvasEnterpriseCustomerConfiguration.DoesNotExist:
+            CanvasEnterpriseCustomerConfiguration.objects.create(
+                uuid=SINGLE_CANVAS_CONFIG['uuid'],
+                decrypted_client_id=SINGLE_CANVAS_CONFIG['client_id'],
+                decrypted_client_secret=SINGLE_CANVAS_CONFIG['client_secret'],
+                canvas_account_id=SINGLE_CANVAS_CONFIG['canvas_account_id'],
+                canvas_base_url=SINGLE_CANVAS_CONFIG['canvas_base_url'],
+                enterprise_customer=self.enterprise_customer,
+                active=True,
+                enterprise_customer_id=ENTERPRISE_ID,
+            )
 
-        CanvasEnterpriseCustomerConfiguration.objects.get_or_create(
-            uuid=SECOND_CANVAS_CONFIG['uuid'],
-            client_id=SECOND_CANVAS_CONFIG['client_id'],
-            client_secret=SECOND_CANVAS_CONFIG['client_secret'],
-            canvas_account_id=SECOND_CANVAS_CONFIG['canvas_account_id'],
-            canvas_base_url=SECOND_CANVAS_CONFIG['canvas_base_url'],
-            enterprise_customer=self.enterprise_customer,
-            active=True,
-            enterprise_customer_id=ENTERPRISE_ID,
-        )
+        try:
+            CanvasEnterpriseCustomerConfiguration.objects.get(
+                uuid=SECOND_CANVAS_CONFIG['uuid'],
+                canvas_account_id=SECOND_CANVAS_CONFIG['canvas_account_id'],
+                canvas_base_url=SECOND_CANVAS_CONFIG['canvas_base_url'],
+                enterprise_customer=self.enterprise_customer,
+                active=True,
+                enterprise_customer_id=ENTERPRISE_ID,
+            )
+        except CanvasEnterpriseCustomerConfiguration.DoesNotExist:
+            CanvasEnterpriseCustomerConfiguration.objects.create(
+                uuid=SECOND_CANVAS_CONFIG['uuid'],
+                decrypted_client_id=SECOND_CANVAS_CONFIG['client_id'],
+                decrypted_client_secret=SECOND_CANVAS_CONFIG['client_secret'],
+                canvas_account_id=SECOND_CANVAS_CONFIG['canvas_account_id'],
+                canvas_base_url=SECOND_CANVAS_CONFIG['canvas_base_url'],
+                enterprise_customer=self.enterprise_customer,
+                active=True,
+                enterprise_customer_id=ENTERPRISE_ID,
+            )
 
     def test_successful_refresh_token_by_uuid_request(self):
         """