fix: LTI 1.3 extra claims and custom parameters (#392)

Allows the LTI consumer XBlock to send custom parameters (including dynamic custom parameters) and extra claims to LTI 1.3 launches.
openedx · Aug 23, 2023 · d3686bf · d3686bf
1 parent 3de9bcb
commit d3686bf
Show file tree

Hide file tree

Showing 7 changed files with 199 additions and 8 deletions.
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -16,6 +16,11 @@ Please See the `releases tab <https://github.com/openedx/xblock-lti-consumer/rel
 Unreleased
 ~~~~~~~~~~
 
+9.6.2 - 2023-08-22
+------------------
+* Fix extra claims and custom parameters for LTI 1.3.
+* Add validation to custom_parameters xblock field.
+
 9.6.1 - 2023-06-28
 ------------------
 * Fix CCX LTI configuration compatibility

diff --git a/lti_consumer/__init__.py b/lti_consumer/__init__.py
@@ -4,4 +4,4 @@
 from .apps import LTIConsumerApp
 from .lti_xblock import LtiConsumerXBlock
 
-__version__ = '9.6.1'
+__version__ = '9.6.2'
diff --git a/lti_consumer/data.py b/lti_consumer/data.py
@@ -83,6 +83,7 @@ class Lti1p3LaunchData:
     * proctoring_launch_data (conditionally required): An instance of the Lti1p3ProctoringLaunchData that contains
         data necessary and related to an LTI 1.3 proctoring launch. It is required if the message_type attribute is
         "LtiStartProctoring" or "LtiEndAssessment".
+    * custom_parameters (optional): The custom parameters claim values. It is a dictionary of custom parameters.
     """
     user_id = field()
     user_role = field()
@@ -104,3 +105,4 @@ class Lti1p3LaunchData:
         default=None,
         validator=validators.optional((validators.instance_of(Lti1p3ProctoringLaunchData))),
     )
+    custom_parameters = field(default={})
diff --git a/lti_consumer/lti_xblock.py b/lti_consumer/lti_xblock.py
@@ -100,7 +100,13 @@
     'staff': 'Administrator',
     'instructor': 'Instructor',
 }
-CUSTOM_PARAMETER_TEMPLATE_TAGS = ('${', '}')
+CUSTOM_PARAMETER_SEPARATOR = '='
+# Allow a key-pair key and value to contain any character except "=".
+CUSTOM_PARAMETER_REGEX = re.compile(
+    rf'^([^{CUSTOM_PARAMETER_SEPARATOR}]+{CUSTOM_PARAMETER_SEPARATOR}[^{CUSTOM_PARAMETER_SEPARATOR}]+)$',
+)
+# Catch a value enclosed by ${}, the value enclosed can contain any charater except "=".
+CUSTOM_PARAMETER_TEMPLATE_REGEX = re.compile(r'^(\${[^%s]+})$' % CUSTOM_PARAMETER_SEPARATOR)
 
 
 def parse_handler_suffix(suffix):
@@ -671,12 +677,20 @@ def _get_statici18n_js_url(loader):  # pragma: no cover
         return None
 
     def validate_field_data(self, validation, data):
+        # Validate custom parameters is a list.
         if not isinstance(data.custom_parameters, list):
             _ = self.runtime.service(self, "i18n").ugettext
             validation.add(ValidationMessage(ValidationMessage.ERROR, str(
                 _("Custom Parameters must be a list")
             )))
 
+        # Validate custom parameters format.
+        if not all(map(CUSTOM_PARAMETER_REGEX.match, data.custom_parameters)):
+            _ = self.runtime.service(self, 'i18n').ugettext
+            validation.add(ValidationMessage(ValidationMessage.ERROR, str(
+                _('Custom Parameters should be strings in "x=y" format.'),
+            )))
+
         # keyset URL and public key are mutually exclusive
         if data.lti_1p3_tool_key_mode == 'keyset_url':
             data.lti_1p3_tool_public_key = ''
@@ -1045,8 +1059,7 @@ def prefixed_custom_parameters(self):
                 if param_name not in LTI_PARAMETERS:
                     param_name = 'custom_' + param_name
 
-                if (param_value.startswith(CUSTOM_PARAMETER_TEMPLATE_TAGS[0]) and
-                        param_value.endswith(CUSTOM_PARAMETER_TEMPLATE_TAGS[1])):
+                if CUSTOM_PARAMETER_TEMPLATE_REGEX.match(param_value):
                     param_value = resolve_custom_parameter_template(self, param_value)
 
                 custom_parameters[param_name] = param_value
@@ -1552,6 +1565,40 @@ def _get_lti_launch_url(self, consumer):
 
         return launch_url
 
+    def get_lti_1p3_custom_parameters(self):
+        """
+        Parses XBlock custom_parameters field values to use on LTI 1.3 launch data.
+        This allows us to transform the list items on the custom_parameters field into a dictionary of
+        custom parameters, it will also resolve any dynamic custom parameter present on the field.
+        Returns:
+            dict: Parsed custom_parameters dictionary
+        """
+        custom_parameters = {}
+
+        # Get site-wide extra parameters from processor functions.
+        for processor in self.get_parameter_processors():
+            try:
+                default_params = getattr(processor, 'lti_xblock_default_params', {})
+                custom_parameters.update(default_params)
+                custom_parameters.update(processor(self) or {})
+            except Exception:  # pylint: disable=broad-except
+                # Log the error without causing a 500-error.
+                # Useful for catching casual runtime errors in the processors.
+                log.exception('Error in XBlock LTI parameter processor "%s"', processor)
+
+        # Get custom parameters from XBlock settings.
+        for parameter in self.custom_parameters:
+            # Split parameter into key, value and remove leading, trailing spaces.
+            key, value = map(str.strip, parameter.split(CUSTOM_PARAMETER_SEPARATOR, 1))
+
+            # Resolve custom parameter template value.
+            if CUSTOM_PARAMETER_TEMPLATE_REGEX.match(value):
+                value = resolve_custom_parameter_template(self, value)
+
+            custom_parameters.update({key: value})
+
+        return custom_parameters
+
     def get_lti_1p3_launch_data(self):
         """
         Return an instance of Lti1p3LaunchData, containing necessary data for an LTI 1.3 launch.
@@ -1595,6 +1642,7 @@ def get_lti_1p3_launch_data(self):
             context_type=["course_offering"],
             context_title=self.get_context_title(),
             context_label=course_key,
+            custom_parameters=self.get_lti_1p3_custom_parameters(),
         )
 
         return launch_data

diff --git a/lti_consumer/plugin/views.py b/lti_consumer/plugin/views.py
@@ -261,6 +261,10 @@ def launch_gate_endpoint(request, suffix=None):  # pylint: disable=unused-argume
         # Set LTI Launch URL.
         context.update({'launch_url': preflight_response.get("redirect_uri")})
 
+        # Set LTI custom properties claim.
+        if launch_data.custom_parameters:
+            lti_consumer.set_custom_parameters(launch_data.custom_parameters)
+
         # Modify LTI Launch URL depending on launch type.
         # Deep Linking Launch - Configuration flow launched by
         # course creators to set up content.

diff --git a/lti_consumer/tests/unit/plugin/test_views.py b/lti_consumer/tests/unit/plugin/test_views.py
@@ -579,6 +579,46 @@ def test_launch_callback_endpoint_end_assessment(self):
         # Check response
         self.assertEqual(response.status_code, 200)
 
+    @patch('lti_consumer.lti_1p3.consumer.LtiConsumer1p3.set_custom_parameters')
+    def test_launch_existing_custom_parameters(self, mock_set_custom_parameters):
+        """
+        Check custom parameters are set if they exist on XBlock configuration.
+        """
+        self.launch_data.custom_parameters = ['test=test']
+        params = {
+            'client_id': self.config.lti_1p3_client_id,
+            'redirect_uri': 'http://tool.example/launch',
+            'state': 'state_test_123',
+            'nonce': 'nonce',
+            'login_hint': self.launch_data.user_id,
+            'lti_message_hint': self.launch_data_key,
+        }
+
+        response = self.client.get(self.url, params)
+
+        self.assertEqual(response.status_code, 200)
+        mock_set_custom_parameters.assert_called_once_with(self.launch_data.custom_parameters)
+
+    @patch('lti_consumer.lti_1p3.consumer.LtiConsumer1p3.set_custom_parameters')
+    def test_launch_non_existing_custom_parameters(self, mock_set_custom_parameters):
+        """
+        Check custom parameters are not set if they don't exist on XBlock configuration.
+        """
+        self.launch_data.custom_parameters = {}
+        params = {
+            'client_id': self.config.lti_1p3_client_id,
+            'redirect_uri': 'http://tool.example/launch',
+            'state': 'state_test_123',
+            'nonce': 'nonce',
+            'login_hint': self.launch_data.user_id,
+            'lti_message_hint': self.launch_data_key,
+        }
+
+        response = self.client.get(self.url, params)
+
+        self.assertEqual(response.status_code, 200)
+        mock_set_custom_parameters.assert_not_called()
+
 
 class TestLti1p3AccessTokenEndpoint(TestCase):
     """

diff --git a/lti_consumer/tests/unit/test_lti_xblock.py b/lti_consumer/tests/unit/test_lti_xblock.py
@@ -15,6 +15,7 @@
 from django.test.testcases import TestCase
 from django.utils import timezone
 from jwkest.jwk import RSAKey, KEYS
+from xblock.validation import Validation
 
 from lti_consumer.exceptions import LtiError
 
@@ -87,6 +88,7 @@ def test_indexibility(self):
         )
 
 
+@ddt.ddt
 class TestProperties(TestLtiConsumerXBlock):
     """
     Unit tests for LtiConsumerXBlock properties
@@ -138,13 +140,48 @@ def test_context_id(self):
         """
         self.assertEqual(self.xblock.context_id, str(self.xblock.scope_ids.usage_id.context_key))
 
-    def test_validate(self):
+    @ddt.data(
+        ['x=y'], [' x x = y y '], ['x= '], [' =y'], [' = '],
+        ['x=y', ' x x = y y ', 'x= ', ' =y', ' = '],
+    )
+    def test_validate_with_valid_custom_parameters(self, custom_parameters):
         """
-        Test that if custom_parameters is empty string, a validation error is added
+        Test if all custom_parameters item are valid
         """
-        self.xblock.custom_parameters = ''
+        self.xblock.custom_parameters = custom_parameters
         validation = self.xblock.validate()
-        self.assertFalse(validation.empty)
+        self.assertEqual(validation.messages, [])
+
+    @patch('lti_consumer.lti_xblock.ValidationMessage')
+    @patch.object(Validation, 'add')
+    def test_validate_with_empty_custom_parameters(self, add_mock, mock_validation_message):
+        """
+        Test if custom_parameters is not a list, a validation error is added
+        """
+        mock_validation_message.ERROR.return_value = 'error'
+        self.xblock.custom_parameters = ''
+
+        self.xblock.validate()
+
+        add_mock.assert_called_once_with(
+            mock_validation_message('error', 'Custom Parameters must be a list'),
+        )
+
+    @ddt.data(['x'], ['x='], ['=y'], ['x==y'], ['x', 'x=', '=y', 'x==y'])
+    @patch('lti_consumer.lti_xblock.ValidationMessage')
+    @patch.object(Validation, 'add')
+    def test_validate_with_invalid_custom_parameters(self, custom_parameters, add_mock, mock_validation_message):
+        """
+        Test if a custom_parameters item is not valid, a validation error is added
+        """
+        mock_validation_message.ERROR.return_value = 'error'
+        self.xblock.custom_parameters = custom_parameters
+
+        self.xblock.validate()
+
+        add_mock.assert_called_once_with(
+            mock_validation_message('error', 'Custom Parameters should be strings in "x=y" format.'),
+        )
 
     @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.course')
     def test_validate_lti_id(self, mock_course):
@@ -1643,6 +1680,56 @@ def setUp(self):
         self.addCleanup(self.mock_filter_enabled_patcher.stop)
         self.addCleanup(self.mock_database_config_enabled_patcher.stop)
 
+    @patch.object(LtiConsumerXBlock, 'get_parameter_processors')
+    @patch('lti_consumer.lti_xblock.resolve_custom_parameter_template')
+    def test_get_lti_1p3_custom_parameters(
+        self,
+        mock_resolve_custom_parameter_template,
+        get_parameter_processors_mock,
+    ):
+        """
+        Test that get_lti_1p3_custom_parameters returns an dictionary of custom parameters
+        """
+        processor_mock = Mock(return_value={'test3': 'test'}, lti_xblock_default_params={})
+        get_parameter_processors_mock.return_value = [processor_mock]
+        mock_resolve_custom_parameter_template.return_value = ''
+        self.xblock.custom_parameters = ['test1=test', 'test2=${test}']
+
+        self.assertDictEqual(
+            self.xblock.get_lti_1p3_custom_parameters(),
+            {'test1': 'test', 'test2': '', 'test3': 'test'},
+        )
+        get_parameter_processors_mock.assert_called_once_with()
+        processor_mock.assert_called_once_with(self.xblock)
+        mock_resolve_custom_parameter_template.assert_called_once_with(self.xblock, '${test}')
+
+    @patch.object(LtiConsumerXBlock, 'get_parameter_processors')
+    @patch('lti_consumer.lti_xblock.resolve_custom_parameter_template')
+    @patch('lti_consumer.lti_xblock.log.exception')
+    def test_get_lti_1p3_custom_parameters_with_invalid_processor(
+        self,
+        log_exception_mock,
+        mock_resolve_custom_parameter_template,
+        get_parameter_processors_mock,
+    ):
+        """
+        Test that get_lti_1p3_custom_parameters logs error when a paramater processor fails.
+        """
+        processor_mock = Mock(side_effect=Exception())
+        get_parameter_processors_mock.return_value = [processor_mock]
+        mock_resolve_custom_parameter_template.return_value = ''
+        self.xblock.custom_parameters = ['test1=test', 'test2=${test}']
+
+        self.assertDictEqual(
+            self.xblock.get_lti_1p3_custom_parameters(),
+            {'test1': 'test', 'test2': ''},
+        )
+        get_parameter_processors_mock.assert_called_once_with()
+        log_exception_mock.assert_called_once_with(
+            'Error in XBlock LTI parameter processor "%s"', processor_mock,
+        )
+        mock_resolve_custom_parameter_template.assert_called_once_with(self.xblock, '${test}')
+
     @ddt.idata(product([True, False], [True, False], [True, False], [True, False]))
     @ddt.unpack
     def test_get_lti_1p3_launch_data(
@@ -1685,6 +1772,9 @@ def test_get_lti_1p3_launch_data(
         # Mock out get_pii_sharing_enabled to reduce the amount of mocking we have to do.
         self.xblock.get_pii_sharing_enabled = Mock(return_value=pii_sharing_enabled)
 
+        # Mock custom_parameters property.
+        self.xblock.get_lti_1p3_custom_parameters = Mock(return_value={})
+
         launch_data = self.xblock.get_lti_1p3_launch_data()
 
         course_key = str(self.xblock.scope_ids.usage_id.course_key)
@@ -1701,6 +1791,7 @@ def test_get_lti_1p3_launch_data(
             "context_type": ["course_offering"],
             "context_title": "context_title",
             "context_label": course_key,
+            'custom_parameters': {},
         }
 
         if pii_sharing_enabled:
@@ -1721,6 +1812,7 @@ def test_get_lti_1p3_launch_data(
             launch_data,
             expected_launch_data
         )
+        self.xblock.get_lti_1p3_custom_parameters.assert_called_once_with()
 
     @patch('lti_consumer.plugin.compat.get_course_by_id')
     def test_get_context_title(self, mock_get_course_by_id):