fix: deep linking target_link_uri and AGS endpoint quirks

.coveragerc

@@ -2,4 +2,5 @@
 [run]
 data_file = .coverage
 source = lti_consumer
-omit = */urls.py
+omit =
+    */urls.py

ISSUE.md

@@ -0,0 +1,39 @@
+# Background
+
+`LtiResourceLinkRequest` and NRPS membership container requires roles claim and must contain at least one role URI from the published role vocabularies.
+
+IMS 1.3 spec lists 3 role vocabularies ([link](https://www.imsglobal.org/spec/lti/v1p3#roles-claim)) but does NOT limit their use to specific cases.
+1. System roles
+2. Institution roles
+3. Context roles
+
+Spec further subdivides each of these into core and non-core URIs and states: `Core roles are those which are most likely to be relevant within LTI and hence vendors should support them by best practice. Vendors may also use the non-core roles, but they may not be widely used`
+
+Table in the image below shows roles supported by some LTI tools vs Open edX.
+
+<img width="930" height="404" alt="Image" src="https://github.com/user-attachments/assets/08492a51-7a33-4c5f-aad8-cfc8f032156a" />
+
+
+Here are some observations from the table above:
+
+1. All tools support context roles which makes sense because a typical LTI launch happens from within a course and therefore the role of the user in that course is most relevant to the launch.
+2. Open edX does NOT support any context roles. This is why integration fails with most of these tools (see [test results](https://openedx.atlassian.net/wiki/spaces/COMM/pages/5985927169/Tools+tested+with+and+status)).
+3.  The institution roles that Open edX supports are really context roles because they are tied to a course and not an org.
+
+The following table shows mapping between role URIs from the spec vs roles in Open edX.
+
+| Course role | Current URIs in roles claim |
+|---|---|
+| Course Admin | `/institution/person#Instructor` (non-core) |
+| Course Staff | `/system/person#Administrator` (core)<br>`/institution/person#Instructor` (non-core) |
+| Limited Staff | `/system/person#Administrator` (core)<br>`/institution/person#Instructor` (non-core) |
+| None | `/institution/person#Student` (core) |
+| All other roles | `/institution/person#Student` (core) |
+
+
+# Proposed way forward
+
+The table below shows current and proposed mapping between Open edX course roles and roles in the spec. Please refer to the spec to get full URIs and other relevant details: https://www.imsglobal.org/spec/lti/v1p3#roles-claim
+
+<img width="1256" height="422" alt="Image" src="https://github.com/user-attachments/assets/4596cd99-68be-40e0-8792-ac751f371ae5" />
+

lti_consumer/admin.py

@@ -7,13 +7,33 @@
 from lti_consumer.forms import CourseAllowPIISharingInLTIAdminForm
 from lti_consumer.models import (
     CourseAllowPIISharingInLTIFlag,
+    Lti1p3Passport,
     LtiAgsLineItem,
     LtiAgsScore,
     LtiConfiguration,
     LtiDlContentItem,
 )
 
 
+class LtiConfigurationInline(admin.TabularInline):
+    """
+    Inline for the LtiConfiguration models in Lti1p3Passport.
+    """
+    model = LtiConfiguration
+    extra = 0
+    can_delete = False
+    fields = ('location',)
+
+    def has_change_permission(self, request, obj=None):  # pragma: nocover
+        return False
+
+    def has_delete_permission(self, request, obj=None):  # pragma: nocover
+        return False
+
+    def has_add_permission(self, request, obj=None):  # pragma: nocover
+        return False
+
+
 @admin.register(LtiConfiguration)
 class LtiConfigurationAdmin(admin.ModelAdmin):
     """
@@ -24,6 +44,20 @@ class LtiConfigurationAdmin(admin.ModelAdmin):
     readonly_fields = ('location', 'config_id')
 
 
+@admin.register(Lti1p3Passport)
+class Lti1p3PassportAdmin(admin.ModelAdmin):
+    """
+    Admin view for Lti1p3Passport models.
+    """
+    list_display = (
+        'name',
+        'context_key',
+        'passport_id',
+        'lti_1p3_client_id',
+    )
+    inlines = [LtiConfigurationInline]
+
+
 @admin.register(CourseAllowPIISharingInLTIFlag)
 class CourseAllowPIISharingInLTIFlagAdmin(KeyedConfigurationModelAdmin):
     """

lti_consumer/api.py

@@ -6,44 +6,95 @@
 """
 
 import json
+import logging
 
 from opaque_keys.edx.keys import CourseKey
 
 from lti_consumer.lti_1p3.constants import LTI_1P3_ROLE_MAP
-from .models import CourseAllowPIISharingInLTIFlag, LtiConfiguration, LtiDlContentItem
+
+from .filters import get_external_config_from_filter
+from .models import CourseAllowPIISharingInLTIFlag, Lti1p3Passport, LtiConfiguration, LtiDlContentItem
 from .utils import (
     get_cache_key,
     get_data_from_cache,
-    get_lti_1p3_context_types_claim,
-    get_lti_deeplinking_content_url,
+    get_lms_lti_access_token_link,
     get_lms_lti_keyset_link,
     get_lms_lti_launch_link,
-    get_lms_lti_access_token_link,
+    get_lti_1p3_context_types_claim,
+    get_lti_deeplinking_content_url,
 )
-from .filters import get_external_config_from_filter
 
+log = logging.getLogger(__name__)
+
+
+def _ensure_lti_passport(block, lti_config):
+    """Ensure passport is synced with block fields, creating a new one if needed."""
+    passport = lti_config.lti_1p3_passport
+    if not passport or lti_config.config_store != LtiConfiguration.CONFIG_ON_XBLOCK:
+        return passport
+
+    block_public_key = str(block.lti_1p3_tool_public_key)
+    block_keyset_url = str(block.lti_1p3_tool_keyset_url)
+
+    # Update existing passport if safe
+    if passport.lticonfiguration_set.count() == 1 or (
+        not passport.lti_1p3_tool_public_key and not passport.lti_1p3_tool_keyset_url
+    ):
+        if passport.lti_1p3_tool_public_key != block_public_key or passport.lti_1p3_tool_keyset_url != block_keyset_url:
+            passport.lti_1p3_tool_public_key = block_public_key
+            passport.lti_1p3_tool_keyset_url = block_keyset_url
+            passport.save()
+            log.info("Updated LTI passport for %s", block.scope_ids.usage_id)
+        return passport
+
+    # Create new passport if keys changed and passport is shared
+    key_mismatch = (
+        block.lti_1p3_tool_key_mode == 'public_key' and passport.lti_1p3_tool_public_key != block_public_key
+    ) or (
+        block.lti_1p3_tool_key_mode == 'keyset_url' and passport.lti_1p3_tool_keyset_url != block_keyset_url
+    )
+
+    if key_mismatch:
+        from lti_consumer.plugin.compat import save_xblock  # pylint: disable=import-outside-toplevel
+        passport = Lti1p3Passport.objects.create(
+            lti_1p3_tool_public_key=block_public_key,
+            lti_1p3_tool_keyset_url=block_keyset_url,
+            name=f"Passport of {block.display_name}",
+            context_key=block.context_id,
+        )
+        block.lti_1p3_passport_id = str(passport.passport_id)
+        save_xblock(block)
+        log.info("Created new LTI passport for %s", block.scope_ids.usage_id)
 
-def _get_or_create_local_lti_config(lti_version, block_location,
-                                    config_store=LtiConfiguration.CONFIG_ON_XBLOCK, external_id=None):
+    return passport
+
+
+def _get_or_create_local_lti_config(lti_version, block, config_store=LtiConfiguration.CONFIG_ON_XBLOCK):
     """
-    Retrieve the LtiConfiguration for the block described by block_location, if one exists. If one does not exist,
-    create an LtiConfiguration with the LtiConfiguration.CONFIG_ON_XBLOCK config_store.
+    Retrieve or create an LtiConfiguration for the block.
 
-    Treat the lti_version argument as the source of truth for LtiConfiguration.version and override the
-    LtiConfiguration.version with lti_version. This allows, for example, for
-    the XBlock to be the source of truth for the LTI version, which is a user-centric perspective we've adopted.
-    This allows XBlock users to update the LTI version without needing to update the database.
+    The lti_version parameter is treated as the source of truth, overriding
+    any stored version to allow XBlocks to control LTI version without DB updates.
     """
-    # The create operation is only performed when there is no existing configuration for the block
-    lti_config, _ = LtiConfiguration.objects.get_or_create(location=block_location)
+    lti_config, _ = LtiConfiguration.objects.get_or_create(location=block.scope_ids.usage_id)
 
-    lti_config.config_store = config_store
-    lti_config.external_id = external_id
+    # Ensure passport is synced with block
+    passport = _ensure_lti_passport(block, lti_config)
 
-    if lti_config.version != lti_version:
-        lti_config.version = lti_version
+    # Batch updates
+    updates = {
+        'config_store': config_store,
+        'external_id': block.external_config,
+        'version': lti_version,
+    }
+    if passport:
+        updates['lti_1p3_passport'] = passport
 
-    lti_config.save()
+    # Only save if changed
+    if any(getattr(lti_config, key) != value for key, value in updates.items()):
+        for key, value in updates.items():
+            setattr(lti_config, key, value)
+        lti_config.save()
 
     return lti_config
 
@@ -65,7 +116,7 @@ def _get_lti_config_for_block(block):
     if block.config_type == 'database':
         lti_config = _get_or_create_local_lti_config(
             block.lti_version,
-            block.scope_ids.usage_id,
+            block,
             LtiConfiguration.CONFIG_ON_DB,
         )
     elif block.config_type == 'external':
@@ -75,14 +126,13 @@ def _get_lti_config_for_block(block):
         )
         lti_config = _get_or_create_local_lti_config(
             config.get("version"),
-            block.scope_ids.usage_id,
+            block,
             LtiConfiguration.CONFIG_EXTERNAL,
-            external_id=block.external_config,
         )
     else:
         lti_config = _get_or_create_local_lti_config(
             block.lti_version,
-            block.scope_ids.usage_id,
+            block,
             LtiConfiguration.CONFIG_ON_XBLOCK,
         )
     return lti_config
@@ -140,7 +190,7 @@ def get_lti_1p3_launch_info(
         if dl_content_items.exists():
             deep_linking_content_items = [item.attributes for item in dl_content_items]
 
-    config_id = lti_config.config_id
+    config_id = lti_config.passport_id
     client_id = lti_config.lti_1p3_client_id
     deployment_id = "1"
 

lti_consumer/lti_1p1/exceptions.py

@@ -1,9 +1,10 @@
 """
 Exceptions for the LTI1.1 Consumer.
 """
+from lti_consumer.exceptions import LtiError
 
 
-class Lti1p1Error(Exception):
+class Lti1p1Error(LtiError):
     """
     General error class for LTI1.1 Consumer usage.
     """

lti_consumer/lti_1p3/consumer.py

@@ -77,6 +77,15 @@ def __init__(
         # Extra claims - used by LTI Advantage
         self.extra_claims = {}
 
+    def _get_target_link_uri(self, launch_data):  # pylint: disable=unused-argument
+        """
+        Return the target_link_uri to use for the provided launch data.
+
+        Subclasses can override this to customize the target link selection for
+        different message types.
+        """
+        return self.launch_url
+
     @staticmethod
     def _get_user_roles(role):
         """
@@ -125,12 +134,14 @@ def prepare_preflight_url(
 
         oidc_url = self.oidc_url + "?"
 
+        target_link_uri = self._get_target_link_uri(launch_data)
+
         login_hint = user_id
         parameters = {
             "iss": self.iss,
             "client_id": self.client_id,
             "lti_deployment_id": self.deployment_id,
-            "target_link_uri": self.launch_url,
+            "target_link_uri": target_link_uri,
             "login_hint": login_hint,
             "lti_message_hint": launch_data_key,
         }
@@ -302,6 +313,7 @@ def set_custom_parameters(
     def get_lti_launch_message(
             self,
             include_extra_claims=True,
+            target_link_uri=None,
     ):
         """
         Build LTI message from class parameters
@@ -312,6 +324,8 @@ def get_lti_launch_message(
         # Start from base message
         lti_message = LTI_BASE_MESSAGE.copy()
 
+        target_link_uri = target_link_uri or self.launch_url
+
         # Add base parameters
         lti_message.update({
             # Issuer
@@ -329,7 +343,7 @@ def get_lti_launch_message(
             # Target Link URI: actual endpoint for the LTI resource to display
             # MUST be the same value as the target_link_uri passed by the platform in the OIDC login request
             # http://www.imsglobal.org/spec/lti/v1p3/#target-link-uri
-            "https://purl.imsglobal.org/spec/lti/claim/target_link_uri": self.launch_url,
+            "https://purl.imsglobal.org/spec/lti/claim/target_link_uri": target_link_uri,
         })
 
         # Check if user data is set, then append it to lti message
@@ -609,6 +623,19 @@ def lti_dl_enabled(self):
         else:
             return False
 
+    def _get_target_link_uri(self, launch_data):
+        """
+        Use the deep linking launch URL for deep linking requests.
+        """
+        if (
+            getattr(self, 'dl', None) and
+            launch_data and
+            getattr(launch_data, 'message_type', None) == 'LtiDeepLinkingRequest'
+        ):
+            return self.dl.deep_linking_launch_url
+
+        return super()._get_target_link_uri(launch_data)
+
     def enable_ags(
         self,
         lineitems_url,
@@ -663,12 +690,14 @@ def generate_launch_request(
 
         # Check if Deep Linking is enabled and that this is a Deep Link Launch
         if self.dl and launch_data.message_type == "LtiDeepLinkingRequest":
+            target_link_uri = self._get_target_link_uri(launch_data)
             # Validate preflight response
             self._validate_preflight_response(preflight_response)
 
             # Get LTI Launch Message
             lti_launch_message = self.get_lti_launch_message(
                 include_extra_claims=False,
+                target_link_uri=target_link_uri,
             )
 
             # Update message type to LtiDeepLinkingRequest,

lti_consumer/lti_1p3/extensions/rest_framework/serializers.py

@@ -125,7 +125,7 @@ class LtiAgsScoreSerializer(serializers.ModelSerializer):
     timestamp = serializers.DateTimeField(input_formats=[ISO_8601], format=ISO_8601, default_timezone=timezone.utc)
     scoreGiven = serializers.FloatField(source='score_given', required=False, allow_null=True, default=None)
     scoreMaximum = serializers.FloatField(source='score_maximum', required=False, allow_null=True, default=None)
-    comment = serializers.CharField(required=False, allow_null=True)
+    comment = serializers.CharField(required=False, allow_null=True, allow_blank=True)
     activityProgress = serializers.CharField(source='activity_progress')
     gradingProgress = serializers.CharField(source='grading_progress')
     userId = serializers.CharField(source='user_id')
@@ -193,14 +193,19 @@ class LtiAgsResultSerializer(serializers.ModelSerializer):
     comment = serializers.CharField()
 
     def get_id(self, obj):
+        """
+        Return result URL for score. Include user_id when score scoped to learner.
+        """
         request = self.context.get('request')
+        kwargs = {
+            'lti_config_id': obj.line_item.lti_configuration.id,
+            'pk': obj.line_item.pk,
+        }
+        if obj.user_id:
+            kwargs['user_id'] = obj.user_id
         return reverse(
             'lti_consumer:lti-ags-view-results',
-            kwargs={
-                'lti_config_id': obj.line_item.lti_configuration.id,
-                'pk': obj.line_item.pk,
-                'user_id': obj.user_id,
-            },
+            kwargs=kwargs,
             request=request,
         )