From f15a78c6054f34411b0a4610083cba91fe5dad7c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9sar=20Montilla?= Date: Wed, 6 Sep 2023 08:45:54 +0000 Subject: [PATCH] feat: Get keyset and access token using external ID fix: Remove multi-tenancy related changes fix: Remove external configuration value overrides --- lti_consumer/api.py | 13 +- .../migrations/0018_add_ags_multi_tenancy.py | 28 ---- lti_consumer/models.py | 122 ++++------------- lti_consumer/plugin/urls.py | 10 ++ lti_consumer/plugin/views.py | 124 +++++++++++++----- lti_consumer/static/js/xblock_studio_view.js | 3 + lti_consumer/tests/unit/plugin/test_views.py | 95 +++++++++++++- .../tests/unit/plugin/test_views_lti_ags.py | 13 -- lti_consumer/tests/unit/test_api.py | 31 ++--- lti_consumer/tests/unit/test_models.py | 46 +------ 10 files changed, 237 insertions(+), 248 deletions(-) delete mode 100644 lti_consumer/migrations/0018_add_ags_multi_tenancy.py diff --git a/lti_consumer/api.py b/lti_consumer/api.py index 0eb8c4c8..569f285d 100644 --- a/lti_consumer/api.py +++ b/lti_consumer/api.py @@ -144,27 +144,20 @@ def get_lti_1p3_launch_info( client_id = lti_config.lti_1p3_client_id token_url = get_lms_lti_access_token_link(config_id) keyset_url = get_lms_lti_keyset_link(config_id) - # We set the deployment ID to a default value of 1, - # this will be used on a configuration with a CONFIG_EXTERNAL config store - # if no deployment ID is set on the external configuration. - deployment_id = '1' # Display LTI launch information from external configuration. # if an external configuration is being used. if lti_config.config_store == lti_config.CONFIG_EXTERNAL: external_config = get_external_config_from_filter({}, lti_config.external_id) client_id = external_config.get('lti_1p3_client_id') - token_url = external_config.get('lti_1p3_access_token_url') - keyset_url = external_config.get('lti_1p3_keyset_url') - # Show default harcoded deployment ID if no deployment ID - # is set on the external configuration. - deployment_id = external_config.get('lti_1p3_deployment_id') or deployment_id + token_url = get_lms_lti_access_token_link(lti_config.external_id.replace(':', '/')) + keyset_url = get_lms_lti_keyset_link(lti_config.external_id.replace(':', '/')) # Return LTI launch information for end user configuration return { 'client_id': client_id, 'keyset_url': keyset_url, - 'deployment_id': deployment_id, + 'deployment_id': '1', 'oidc_callback': get_lms_lti_launch_link(), 'token_url': token_url, 'deep_linking_launch_url': deep_linking_launch_url, diff --git a/lti_consumer/migrations/0018_add_ags_multi_tenancy.py b/lti_consumer/migrations/0018_add_ags_multi_tenancy.py deleted file mode 100644 index a403f2ab..00000000 --- a/lti_consumer/migrations/0018_add_ags_multi_tenancy.py +++ /dev/null @@ -1,28 +0,0 @@ -# Generated by Django 3.2.17 on 2023-07-07 02:57 - -from django.db import migrations, models - - -class Migration(migrations.Migration): - - dependencies = [ - ('lti_consumer', '0017_lticonfiguration_lti_1p3_redirect_uris'), - ] - - operations = [ - migrations.AddField( - model_name='ltiagslineitem', - name='client_id', - field=models.CharField(blank=True, max_length=255, null=True), - ), - migrations.AddField( - model_name='ltiagslineitem', - name='deployment_id', - field=models.CharField(blank=True, max_length=255, null=True), - ), - migrations.AddField( - model_name='ltiagslineitem', - name='oidc_url', - field=models.CharField(blank=True, max_length=255, null=True), - ), - ] diff --git a/lti_consumer/models.py b/lti_consumer/models.py index 5d5f15a3..09572bae 100644 --- a/lti_consumer/models.py +++ b/lti_consumer/models.py @@ -412,14 +412,8 @@ def get_lti_advantage_deep_linking_launch_url(self): if self.config_store == self.CONFIG_ON_DB: return self.lti_advantage_deep_linking_launch_url elif self.config_store == self.CONFIG_EXTERNAL: - block = compat.load_enough_xblock(self.location) config = get_external_config_from_filter({}, self.external_id) - return ( - # Use the deep linking launch URL set in XBlock or DB has a fallback. - block.lti_advantage_deep_linking_launch_url or - self.lti_advantage_deep_linking_launch_url or - config.get('lti_advantage_deep_linking_launch_url') - ) + return config.get('lti_advantage_deep_linking_launch_url') else: block = compat.load_enough_xblock(self.location) return block.lti_advantage_deep_linking_launch_url @@ -449,39 +443,7 @@ def _setup_lti_1p3_ags(self, consumer): log.info('LTI Advantage AGS is disabled for %s', self) return - # We set the deployment ID to a default value of 1, - # this will be used on a configuration with a CONFIG_EXTERNAL config store - # if no deployment ID is set on the external configuration. - deployment_id = '1' - - # Get OIDC URL, Client ID and Deployment ID (On CONFIG_EXTERNAL) - # to use it to retrieve/save the tool deployment LineItem. - if self.config_store == self.CONFIG_ON_XBLOCK: - block = compat.load_enough_xblock(self.location) - oidc_url = block.lti_1p3_oidc_url - client_id = block.lti_1p3_client_id - elif self.config_store == self.CONFIG_EXTERNAL: - block = compat.load_enough_xblock(self.location) - config = get_external_config_from_filter({}, self.external_id) - # Use the OIDC URL set in XBlock or DB has a fallback. - oidc_url = ( - block.lti_1p3_oidc_url or - self.lti_1p3_oidc_url or - config.get('lti_1p3_oidc_url') - ) - client_id = config.get('lti_1p3_client_id') - # Deployment ID hardcoded to 1 if no deployment ID is set - # on the external configuration. - deployment_id = config.get('lti_1p3_deployment_id') or deployment_id - else: - oidc_url = self.lti_1p3_oidc_url - client_id = self.lti_1p3_client_id - - lineitem = self.ltiagslineitem_set.filter( - oidc_url=oidc_url, - client_id=client_id, - deployment_id=deployment_id, - ).first() + lineitem = self.ltiagslineitem_set.first() # If using the declarative approach, we should create a LineItem if it # doesn't exist. This is because on this mode the tool is not able to create @@ -515,9 +477,6 @@ def _setup_lti_1p3_ags(self, consumer): lineitem = LtiAgsLineItem.objects.create( lti_configuration=self, resource_link_id=self.location, - oidc_url=oidc_url, - client_id=client_id, - deployment_id=deployment_id, **default_values ) @@ -606,33 +565,22 @@ def _get_lti_1p3_consumer(self): tool_keyset_url=self.lti_1p3_tool_keyset_url, ) elif self.config_store == self.CONFIG_EXTERNAL: - block = compat.load_enough_xblock(self.location) - external = get_external_config_from_filter({}, self.external_id) + config = get_external_config_from_filter({}, self.external_id) consumer = consumer_class( iss=get_lti_api_base(), - # Use the OIDC URL set in XBlock or DB has a fallback. - lti_oidc_url=( - block.lti_1p3_oidc_url or - self.lti_1p3_oidc_url or - external.get('lti_1p3_oidc_url') - ), - # Use the launch URL set in XBlock or DB has a fallback. - lti_launch_url=( - block.lti_1p3_launch_url or - self.lti_1p3_launch_url or - external.get('lti_1p3_launch_url') - ), - client_id=external.get('lti_1p3_client_id'), - # Deployment ID hardcoded to 1 if no deployment ID is set - # on the external configuration. - deployment_id=external.get('lti_1p3_deployment_id') or '1', - rsa_key=external.get('lti_1p3_private_key'), - rsa_key_id=external.get('lti_1p3_private_key_id'), + lti_oidc_url=config.get('lti_1p3_oidc_url'), + lti_launch_url=config.get('lti_1p3_launch_url'), + client_id=config.get('lti_1p3_client_id'), + # Deployment ID hardcoded to 1 since + # we're not using multi-tenancy. + deployment_id='1', + rsa_key=config.get('lti_1p3_private_key'), + rsa_key_id=config.get('lti_1p3_private_key_id'), # Registered redirect uris redirect_uris=self.get_lti_1p3_redirect_uris(), - tool_key=external.get('lti_1p3_tool_public_key'), - tool_keyset_url=external.get('lti_1p3_tool_keyset_url'), + tool_key=config.get('lti_1p3_tool_public_key'), + tool_keyset_url=config.get('lti_1p3_tool_keyset_url'), ) else: # This should not occur, but raise an error if self.config_store is not @@ -660,23 +608,10 @@ def get_lti_1p3_redirect_uris(self): Return pre-registered redirect uris or sensible defaults """ if self.config_store == self.CONFIG_EXTERNAL: - block = compat.load_enough_xblock(self.location) config = get_external_config_from_filter({}, self.external_id) - redirect_uris = ( - block.lti_1p3_redirect_uris or - self.lti_1p3_redirect_uris or - config.get('lti_1p3_redirect_uris') - ) - launch_url = ( - block.lti_1p3_launch_url or - self.lti_1p3_launch_url or - config.get('lti_1p3_launch_url') - ) - deep_link_launch_url = ( - block.lti_advantage_deep_linking_launch_url or - self.lti_advantage_deep_linking_launch_url or - config.get('lti_advantage_deep_linking_launch_url') - ) + redirect_uris = config.get('lti_1p3_redirect_uris') + launch_url = config.get('lti_1p3_launch_url') + deep_link_launch_url = config.get('lti_advantage_deep_linking_launch_url') elif self.config_store == self.CONFIG_ON_DB: redirect_uris = self.lti_1p3_redirect_uris launch_url = self.lti_1p3_launch_url @@ -723,6 +658,10 @@ class LtiAgsLineItem(models.Model): LTI-AGS Specification: https://www.imsglobal.org/spec/lti-ags/v2p0 The platform MUST NOT modify the 'resourceId', 'resourceLinkId' and 'tag' values. + Note: When implementing multi-tenancy support, this needs to be changed + and be tied to a deployment ID, because each deployment should isolate + it's resources. + .. no_pii: """ # LTI Configuration link @@ -736,24 +675,6 @@ class LtiAgsLineItem(models.Model): blank=True ) - # OIDC URL, Client ID and Deployment ID - # This allows us to isolate the LineItem per tool deployment - oidc_url = models.CharField( - max_length=255, - blank=True, - null=True, - ) - client_id = models.CharField( - max_length=255, - blank=True, - null=True, - ) - deployment_id = models.CharField( - max_length=255, - blank=True, - null=True, - ) - # Tool resource identifier, not used by the LMS. resource_id = models.CharField(max_length=100, blank=True) @@ -791,6 +712,9 @@ class LtiAgsScore(models.Model): Model to store LineItem Score data for LTI Assignments and Grades service. LTI-AGS Specification: https://www.imsglobal.org/spec/lti-ags/v2p0 + Note: When implementing multi-tenancy support, this needs to be changed + and be tied to a deployment ID, because each deployment should isolate + it's resources. .. no_pii: """ diff --git a/lti_consumer/plugin/urls.py b/lti_consumer/plugin/urls.py index 69780718..d53f570b 100644 --- a/lti_consumer/plugin/urls.py +++ b/lti_consumer/plugin/urls.py @@ -31,6 +31,11 @@ public_keyset_endpoint, name='lti_consumer.public_keyset_endpoint_via_location' ), + path( + 'lti_consumer/v1/public_keysets//', + public_keyset_endpoint, + name='lti_consumer.public_keyset_endpoint_via_external_id' + ), re_path( 'lti_consumer/v1/launch/(?:/(?P.*))?$', launch_gate_endpoint, @@ -46,6 +51,11 @@ access_token_endpoint, name='lti_consumer.access_token_via_location' ), + path( + 'lti_consumer/v1/token//', + access_token_endpoint, + name='lti_consumer.access_token_via_external_id' + ), re_path( r'lti_consumer/v1/lti/(?P[-\w]+)/lti-dl/response', deep_linking_response_endpoint, diff --git a/lti_consumer/plugin/views.py b/lti_consumer/plugin/views.py index 15f71e2f..bb30244f 100644 --- a/lti_consumer/plugin/views.py +++ b/lti_consumer/plugin/views.py @@ -6,7 +6,7 @@ from django.conf import settings from django.contrib.auth import get_user_model -from django.core.exceptions import ObjectDoesNotExist, PermissionDenied, ValidationError +from django.core.exceptions import PermissionDenied, ValidationError from django.db import transaction from django.http import Http404, JsonResponse from django.shortcuts import render @@ -26,7 +26,7 @@ from lti_consumer.api import get_lti_pii_sharing_state_for_course, validate_lti_1p3_launch_data from lti_consumer.exceptions import LtiError -from lti_consumer.lti_1p3.consumer import LtiProctoringConsumer +from lti_consumer.lti_1p3.consumer import LtiConsumer1p3, LtiProctoringConsumer from lti_consumer.lti_1p3.exceptions import (BadJwtSignature, InvalidClaimValue, Lti1p3Exception, LtiDeepLinkingContentTypeNotSupported, MalformedJwtToken, MissingRequiredClaim, NoSuitableKeys, TokenSignatureExpired, @@ -48,7 +48,8 @@ from lti_consumer.plugin import compat from lti_consumer.signals.signals import LTI_1P3_PROCTORING_ASSESSMENT_STARTED from lti_consumer.track import track_event -from lti_consumer.utils import _, get_data_from_cache, get_lti_1p3_context_types_claim +from lti_consumer.utils import _, get_data_from_cache, get_lti_1p3_context_types_claim, get_lti_api_base +from lti_consumer.filters import get_external_config_from_filter log = logging.getLogger(__name__) @@ -83,21 +84,50 @@ def has_block_access(user, block, course_key): @require_http_methods(["GET"]) -def public_keyset_endpoint(request, usage_id=None, lti_config_id=None): +def public_keyset_endpoint( + request, + usage_id=None, + lti_config_id=None, + external_app=None, + external_slug=None, +): """ Gate endpoint to fetch public keysets from a problem This is basically a passthrough function that uses the OIDC response parameter `login_hint` to locate the block and run the proper handler. + + Arguments: + lti_config_id (UUID): config_id of the LtiConfiguration + usage_id (UsageKey): location of the Block + external_app (str): App name of the external LTI configuration + external_slug (str): Slug of the external LTI configuration. + + Returns: + JsonResponse or Http404 """ + external_id = f"{external_app}:{external_slug}" + try: if usage_id: lti_config = LtiConfiguration.objects.get(location=UsageKey.from_string(usage_id)) + version = lti_config.version + public_jwk = lti_config.lti_1p3_public_jwk elif lti_config_id: lti_config = LtiConfiguration.objects.get(config_id=lti_config_id) + version = lti_config.version + public_jwk = lti_config.lti_1p3_public_jwk + elif external_app and external_slug: + lti_config = get_external_config_from_filter({}, external_id) + + if not lti_config: + raise ValueError("External LTI configuration not found") - if lti_config.version != lti_config.LTI_1P3: + version = lti_config.get("version") + public_jwk = lti_config.get("lti_1p3_public_jwk", {}) + + if version != LtiConfiguration.LTI_1P3: raise LtiError( "LTI Error: LTI 1.1 blocks do not have a public keyset endpoint." ) @@ -105,14 +135,13 @@ def public_keyset_endpoint(request, usage_id=None, lti_config_id=None): # Retrieve block's Public JWK # The underlying method will generate a new Private-Public Pair if one does # not exist, and retrieve the values. - response = JsonResponse(lti_config.lti_1p3_public_jwk) + response = JsonResponse(public_jwk) response['Content-Disposition'] = 'attachment; filename=keyset.json' return response - except (LtiError, InvalidKeyError, ObjectDoesNotExist) as exc: + except (InvalidKeyError, LtiConfiguration.DoesNotExist, ValueError, LtiError) as exc: log.info( - "Error while retrieving keyset for usage_id (%r) or lit_config_id (%s): %s", - usage_id, - lti_config_id, + "Error while retrieving keyset for ID %s: %s", + usage_id or lti_config_id or external_id, exc, exc_info=True, ) @@ -362,7 +391,13 @@ def launch_gate_endpoint(request, suffix=None): # pylint: disable=unused-argume @csrf_exempt @xframe_options_sameorigin @require_http_methods(["POST"]) -def access_token_endpoint(request, lti_config_id=None, usage_id=None): +def access_token_endpoint( + request, + lti_config_id=None, + usage_id=None, + external_app=None, + external_slug=None, +): """ Gate endpoint to enable tools to retrieve access tokens for the LTI 1.3 tool. @@ -371,6 +406,8 @@ def access_token_endpoint(request, lti_config_id=None, usage_id=None): Arguments: lti_config_id (UUID): config_id of the LtiConfiguration usage_id (UsageKey): location of the Block + external_app (str): App name of the external LTI configuration + external_slug (str): Slug of the external LTI configuration. Returns: JsonResponse or Http404 @@ -379,21 +416,50 @@ def access_token_endpoint(request, lti_config_id=None, usage_id=None): Sucess: https://tools.ietf.org/html/rfc6749#section-4.4.3 Failure: https://tools.ietf.org/html/rfc6749#section-5.2 """ + external_id = f"{external_app}:{external_slug}" try: - if lti_config_id: + if usage_id: + lti_config = LtiConfiguration.objects.get(location=UsageKey.from_string(usage_id)) + version = lti_config.version + lti_consumer = lti_config.get_lti_consumer() + elif lti_config_id: lti_config = LtiConfiguration.objects.get(config_id=lti_config_id) - else: - usage_key = UsageKey.from_string(usage_id) - lti_config = LtiConfiguration.objects.get(location=usage_key) - except LtiConfiguration.DoesNotExist as exc: - log.warning("Error getting the LTI configuration with id %r: %s", lti_config_id, exc, exc_info=True) + version = lti_config.version + lti_consumer = lti_config.get_lti_consumer() + elif external_app and external_slug: + lti_config = get_external_config_from_filter({}, external_id) + + if not lti_config: + raise ValueError("External LTI configuration not found") + + version = lti_config.get("version") + # External LTI configurations don't have a get_lti_consumer method + # so we initialize the LtiConsumer1p3 class using the external config data. + lti_consumer = LtiConsumer1p3( + iss=get_lti_api_base(), + lti_oidc_url=None, + lti_launch_url=None, + client_id=lti_config.get("lti_1p3_client_id"), + deployment_id=None, + rsa_key=lti_config.get("lti_1p3_private_key"), + rsa_key_id=lti_config.get("lti_1p3_private_key_id"), + redirect_uris=None, + tool_key=lti_config.get("lti_1p3_tool_public_key"), + tool_keyset_url=lti_config.get("lti_1p3_tool_keyset_url"), + ) + except (InvalidKeyError, LtiConfiguration.DoesNotExist, ValueError) as exc: + log.info( + "Error while retrieving access token for ID %s: %s", + usage_id or lti_config_id or external_id, + exc, + exc_info=True, + ) raise Http404 from exc - if lti_config.version != lti_config.LTI_1P3: + if version != LtiConfiguration.LTI_1P3: return JsonResponse({"error": "invalid_lti_version"}, status=HTTP_404_NOT_FOUND) - lti_consumer = lti_config.get_lti_consumer() try: token = lti_consumer.access_token( dict(urllib.parse.parse_qsl( @@ -606,25 +672,19 @@ class LtiAgsLineItemViewset(viewsets.ModelViewSet): def get_queryset(self): lti_configuration = self.request.lti_configuration - consumer_dict = lti_configuration.get_lti_consumer().__dict__ + # Return all LineItems related to the LTI configuration. + # TODO: + # Note that each configuration currently maps 1:1 + # to each resource link (block), and this filter needs + # improved once we start reusing LTI configurations. return LtiAgsLineItem.objects.filter( - lti_configuration=lti_configuration, - oidc_url=consumer_dict.get('oidc_url'), - client_id=consumer_dict.get('client_id'), - deployment_id=consumer_dict.get('deployment_id'), + lti_configuration=lti_configuration ) def perform_create(self, serializer): lti_configuration = self.request.lti_configuration - consumer_dict = lti_configuration.get_lti_consumer().__dict__ - - serializer.save( - lti_configuration=lti_configuration, - oidc_url=consumer_dict.get('oidc_url'), - client_id=consumer_dict.get('client_id'), - deployment_id=consumer_dict.get('deployment_id'), - ) + serializer.save(lti_configuration=lti_configuration) @action( detail=True, diff --git a/lti_consumer/static/js/xblock_studio_view.js b/lti_consumer/static/js/xblock_studio_view.js index 488872f7..bece55a6 100644 --- a/lti_consumer/static/js/xblock_studio_view.js +++ b/lti_consumer/static/js/xblock_studio_view.js @@ -78,9 +78,12 @@ function LtiConsumerXBlockInitStudio(runtime, element) { const configType = $(element).find('#xb-field-edit-config_type').val(); const databaseConfigHiddenFields = lti1P1FieldList.concat(lti1P3FieldList); const externalConfigHiddenFields = lti1P1FieldList.concat([ + 'lti_1p3_oidc_url', + 'lti_1p3_launch_url', 'lti_1p3_tool_key_mode', 'lti_1p3_tool_keyset_url', 'lti_1p3_tool_public_key', + 'lti_advantage_deep_linking_launch_url', ]); const fieldsToHide = []; diff --git a/lti_consumer/tests/unit/plugin/test_views.py b/lti_consumer/tests/unit/plugin/test_views.py index a2c701a2..7c62ce1a 100644 --- a/lti_consumer/tests/unit/plugin/test_views.py +++ b/lti_consumer/tests/unit/plugin/test_views.py @@ -30,6 +30,7 @@ from lti_consumer.utils import cache_lti_1p3_launch_data +@ddt.ddt class TestLti1p3KeysetEndpoint(TestCase): """ Test `public_keyset_endpoint` method. @@ -74,6 +75,19 @@ def test_invalid_usage_key(self): response = self.client.get('/lti_consumer/v1/public_keysets/invalid-key') self.assertEqual(response.status_code, 404) + @ddt.data( + ('lti_consumer:lti_consumer.public_keyset_endpoint', ['075194d3-6885-417e-a8a8-6c931e272f00']), + ('lti_consumer:lti_consumer.public_keyset_endpoint_via_location', ['block-v1:x+x+x+type@problem+block@x']), + ('lti_consumer:lti_consumer.public_keyset_endpoint_via_external_id', ['x', 'x']), + ) + @ddt.unpack + def test_non_existant_configuration(self, url, args): + """ + Check that 404 is returned when there is no configuration found. + """ + response = self.client.get(reverse(url, args=args)) + self.assertEqual(response.status_code, 404) + def test_wrong_lti_version(self): """ Check if trying to fetch the public keyset for LTI 1.1 yields a HTTP code 404. @@ -102,6 +116,25 @@ def test_public_keyset_endpoint_using_lti_config_id_in_url(self): json.loads(response.content.decode('utf-8')) ) + @patch('lti_consumer.plugin.views.get_external_config_from_filter') + def test_public_keyset_endpoint_using_external_id_in_url(self, get_external_config_from_filter): + """ + Check that the endpoint is accessible using the external LTI configuration ID. + """ + external_config = {'version': LtiConfiguration.LTI_1P3, 'lti_1p3_public_jwk': {}} + get_external_config_from_filter.return_value = external_config + + response = self.client.get(reverse( + 'lti_consumer:lti_consumer.public_keyset_endpoint_via_external_id', + args=['x', 'x'] + )) + + get_external_config_from_filter.assert_called_once_with({}, "x:x") + self.assertEqual(response.status_code, 200) + self.assertEqual(response['Content-type'], 'application/json') + self.assertEqual(response['Content-Disposition'], 'attachment; filename=keyset.json') + self.assertEqual(external_config["lti_1p3_public_jwk"], json.loads(response.content.decode('utf-8'))) + @ddt.ddt class TestLti1p3LaunchGateEndpoint(TestCase): @@ -620,6 +653,7 @@ def test_launch_non_existing_custom_parameters(self, mock_set_custom_parameters) mock_set_custom_parameters.assert_not_called() +@ddt.ddt class TestLti1p3AccessTokenEndpoint(TestCase): """ Test `access_token_endpoint` method. @@ -687,12 +721,65 @@ def test_access_token_endpoint_with_location_in_url(self): self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), token) - def test_non_existant_configuration_for_given_id(self): + @patch('lti_consumer.plugin.views.get_lti_api_base') + @patch('lti_consumer.plugin.views.LtiConsumer1p3') + @patch('lti_consumer.plugin.views.get_external_config_from_filter') + def test_access_token_endpoint_with_external_id_in_url( + self, + get_external_config_from_filter, + lti_consumer, + get_lti_api_base, + ): + """ + Check that the access_token generated by the lti_consumer is returned. + """ + external_config = { + 'version': LtiConfiguration.LTI_1P3, + 'lti_1p3_client_id': 'client-id', + 'lti_1p3_private_key': 'private-key', + 'lti_1p3_private_key_id': 'private-key-id', + 'lti_1p3_tool_public_key': 'public-key', + 'lti_1p3_tool_keyset_url': 'keyset-url', + } + token = {'access_token': 'test-token'} + get_external_config_from_filter.return_value = external_config + lti_consumer.return_value.access_token.return_value = token + + body = self.get_body(create_jwt(self.key, {})) + response = self.client.post( + reverse('lti_consumer:lti_consumer.access_token_via_external_id', args=['x', 'x']), + data=body, + ) + + get_external_config_from_filter.assert_called_once_with({}, 'x:x') + get_lti_api_base.assert_called_once_with() + lti_consumer.assert_called_once_with( + iss=get_lti_api_base(), + lti_oidc_url=None, + lti_launch_url=None, + client_id=external_config['lti_1p3_client_id'], + deployment_id=None, + rsa_key=external_config['lti_1p3_private_key'], + rsa_key_id=external_config['lti_1p3_private_key_id'], + redirect_uris=None, + tool_key=external_config['lti_1p3_tool_public_key'], + tool_keyset_url=external_config['lti_1p3_tool_keyset_url'], + ) + lti_consumer().access_token.called_once_with(body) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), token) + + @ddt.data( + ('lti_consumer:lti_consumer.access_token', ['075194d3-6885-417e-a8a8-6c931e272f00']), + ('lti_consumer:lti_consumer.access_token_via_location', ['block-v1:x+x+x+type@problem+block@x']), + ('lti_consumer:lti_consumer.access_token_via_external_id', ['x', 'x']), + ) + @ddt.unpack + def test_non_existant_configuration(self, url, args): """ - Check that 404 is returned when there is no configuration for a given id + Check that 404 is returned when there is no configuration found. """ - url = reverse('lti_consumer:lti_consumer.access_token', args=['075194d3-6885-417e-a8a8-6c931e272f00']) - response = self.client.post(url) + response = self.client.post(reverse(url, args=args)) self.assertEqual(response.status_code, 404) def test_verify_lti_version_is_1p3(self): diff --git a/lti_consumer/tests/unit/plugin/test_views_lti_ags.py b/lti_consumer/tests/unit/plugin/test_views_lti_ags.py index 825b7b62..7a9e850f 100644 --- a/lti_consumer/tests/unit/plugin/test_views_lti_ags.py +++ b/lti_consumer/tests/unit/plugin/test_views_lti_ags.py @@ -69,7 +69,6 @@ def setUp(self): self._compat_mock = compat_mock.start() self._compat_mock.get_user_from_external_user_id.return_value = self._mock_user self._compat_mock.load_block_as_user.return_value = self.xblock - self.consumer_dict = self.lti_config.get_lti_consumer().__dict__ def _set_lti_token(self, scopes=None): """ @@ -171,9 +170,6 @@ def test_lti_ags_list(self): # Create LineItem line_item = LtiAgsLineItem.objects.create( lti_configuration=self.lti_config, - oidc_url=self.consumer_dict.get('oidc_url'), - client_id=self.consumer_dict.get('client_id'), - deployment_id=self.consumer_dict.get('deployment_id'), resource_id="test", resource_link_id=self.xblock.scope_ids.usage_id, label="test label", @@ -212,9 +208,6 @@ def test_lti_ags_retrieve(self): # Create LineItem line_item = LtiAgsLineItem.objects.create( lti_configuration=self.lti_config, - oidc_url=self.consumer_dict.get('oidc_url'), - client_id=self.consumer_dict.get('client_id'), - deployment_id=self.consumer_dict.get('deployment_id'), resource_id="test", resource_link_id=self.xblock.scope_ids.usage_id, label="test label", @@ -324,9 +317,6 @@ def setUp(self): # Create LineItem self.line_item = LtiAgsLineItem.objects.create( lti_configuration=self.lti_config, - oidc_url=self.consumer_dict.get('oidc_url'), - client_id=self.consumer_dict.get('client_id'), - deployment_id=self.consumer_dict.get('deployment_id'), resource_id="test", resource_link_id=self.xblock.scope_ids.usage_id, label="test label", @@ -859,9 +849,6 @@ def setUp(self): # Create LineItem self.line_item = LtiAgsLineItem.objects.create( lti_configuration=self.lti_config, - oidc_url=self.consumer_dict.get('oidc_url'), - client_id=self.consumer_dict.get('client_id'), - deployment_id=self.consumer_dict.get('deployment_id'), resource_id="test", resource_link_id=self.xblock.scope_ids.usage_id, label="test label", diff --git a/lti_consumer/tests/unit/test_api.py b/lti_consumer/tests/unit/test_api.py index 8e935b6d..5b3095a1 100644 --- a/lti_consumer/tests/unit/test_api.py +++ b/lti_consumer/tests/unit/test_api.py @@ -416,7 +416,6 @@ def test_required_start_assessment_url_for_start_proctoring_message_type(self, s ) -@ddt.ddt class TestGetLti1p3LaunchInfo(TestCase): """ Unit tests for get_lti_1p3_launch_info API method. @@ -539,35 +538,23 @@ def test_launch_info_for_lti_config_without_location(self): } ) - @ddt.data( - { - 'lti_1p3_client_id': 'test-client-id', - 'lti_1p3_access_token_url': 'test-access-token-url', - 'lti_1p3_keyset_url': 'test-keyset-url', - 'lti_1p3_deployment_id': 'test-deployment-id', - }, - { - 'lti_1p3_client_id': 'test-client-id', - 'lti_1p3_access_token_url': 'test-access-token-url', - 'lti_1p3_keyset_url': 'test-keyset-url', - }, - ) @patch('lti_consumer.api.get_external_config_from_filter') def test_launch_info_for_lti_config_with_external_configuration( self, - external_config, filter_mock, ): """ Check if the API can return launch info for LtiConfiguration using an external configuration. """ + external_id = 'test-app:test-slug' + external_config = {'lti_1p3_client_id': 'test-client-id'} filter_mock.return_value = external_config - LtiConfiguration.objects.create( + lti_config = LtiConfiguration.objects.create( version=LtiConfiguration.LTI_1P3, config_id=_test_config_id, config_store=LtiConfiguration.CONFIG_EXTERNAL, - external_id='test-external-id', + external_id=external_id, ) launch_info = get_lti_1p3_launch_info(self._get_lti_1p3_launch_data()) @@ -576,15 +563,19 @@ def test_launch_info_for_lti_config_with_external_configuration( launch_info, { 'client_id': external_config.get('lti_1p3_client_id'), - 'keyset_url': external_config.get('lti_1p3_keyset_url'), + 'keyset_url': 'https://example.com/api/lti_consumer/v1/public_keysets/{}'.format( + lti_config.external_id.replace(':', '/'), + ), 'deployment_id': external_config.get('lti_1p3_deployment_id', '1'), 'oidc_callback': 'https://example.com/api/lti_consumer/v1/launch/', - 'token_url': external_config.get('lti_1p3_access_token_url'), + 'token_url': 'https://example.com/api/lti_consumer/v1/token/{}'.format( + lti_config.external_id.replace(':', '/'), + ), 'deep_linking_launch_url': 'http://example.com', 'deep_linking_content_items': None, }, ) - filter_mock.assert_called_once_with({}, 'test-external-id') + filter_mock.assert_called_once_with({}, external_id) class TestGetLti1p3LaunchUrl(Lti1P3TestCase): diff --git a/lti_consumer/tests/unit/test_models.py b/lti_consumer/tests/unit/test_models.py index bd2accee..011679ae 100644 --- a/lti_consumer/tests/unit/test_models.py +++ b/lti_consumer/tests/unit/test_models.py @@ -279,13 +279,15 @@ def test_get_lti_advantage_deep_linking_enabled(self, config_store, expected_val @ddt.data( {'config_store': LtiConfiguration.CONFIG_ON_XBLOCK, 'expected_value': 'XBlock'}, {'config_store': LtiConfiguration.CONFIG_ON_DB, 'expected_value': 'database'}, - {'config_store': LtiConfiguration.CONFIG_EXTERNAL, 'expected_value': 'XBlock'}, + {'config_store': LtiConfiguration.CONFIG_EXTERNAL, 'expected_value': 'external'}, ) @ddt.unpack - def test_get_lti_advantage_deep_linking_launch_url(self, config_store, expected_value): + @patch('lti_consumer.models.get_external_config_from_filter') + def test_get_lti_advantage_deep_linking_launch_url(self, filter_mock, config_store, expected_value): """ Check if LTI Deep Linking launch URL is properly returned. """ + filter_mock.return_value = {'lti_advantage_deep_linking_launch_url': 'external'} config = self._get_1p3_config(config_store=config_store, lti_advantage_deep_linking_launch_url='database') self.xblock.lti_advantage_deep_linking_launch_url = 'XBlock' @@ -430,46 +432,6 @@ def test_get_redirect_uris_with_external_config( external_config['lti_advantage_deep_linking_launch_url'], ) - @patch('lti_consumer.models.choose_lti_1p3_redirect_uris', return_value=None) - @patch('lti_consumer.models.get_external_config_from_filter') - def test_get_redirect_uris_with_external_config_and_xblock_override( - self, - get_external_config_from_filter_mock, - choose_lti_1p3_redirect_uris, - ): - """ - Test get_redirect_uris with external configuration and XBlock override. - """ - redirect_uris = ['xblock-redirect-uris'] - get_external_config_from_filter_mock.return_value = {} - self.xblock.lti_1p3_redirect_uris = redirect_uris - self.xblock.lti_1p3_launch_url = LAUNCH_URL - self.xblock.lti_advantage_deep_linking_launch_url = DEEP_LINK_URL - - self.assertEqual(self.lti_1p3_config_external.get_lti_1p3_redirect_uris(), None) - get_external_config_from_filter_mock.assert_called_once_with({}, self.lti_1p3_config_external.external_id) - choose_lti_1p3_redirect_uris.assert_called_once_with(redirect_uris, LAUNCH_URL, DEEP_LINK_URL) - - @patch('lti_consumer.models.choose_lti_1p3_redirect_uris', return_value=None) - @patch('lti_consumer.models.get_external_config_from_filter') - def test_get_redirect_uris_with_external_config_and_db_override( - self, - get_external_config_from_filter_mock, - choose_lti_1p3_redirect_uris, - ): - """ - Test get_redirect_uris with external configuration and DB override. - """ - redirect_uris = ['db-redirect-uris'] - get_external_config_from_filter_mock.return_value = {} - self.lti_1p3_config_external.lti_1p3_redirect_uris = redirect_uris - self.lti_1p3_config_external.lti_1p3_launch_url = LAUNCH_URL - self.lti_1p3_config_external.lti_advantage_deep_linking_launch_url = DEEP_LINK_URL - - self.assertEqual(self.lti_1p3_config_external.get_lti_1p3_redirect_uris(), None) - get_external_config_from_filter_mock.assert_called_once_with({}, self.lti_1p3_config_external.external_id) - choose_lti_1p3_redirect_uris.assert_called_once_with(redirect_uris, LAUNCH_URL, DEEP_LINK_URL) - @patch.object(LtiConfiguration, 'sync_configurations') def test_save(self, sync_configurations_mock): """Test save method."""