diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 9a22b279addb7800b33fcffccae3ab4f3701d612..34832279037e0e872873df2158fa7ee4b9eb989d 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -15,14 +15,28 @@ Please See the [releases tab](https://github.com/edx/xblock-lti-consumer/release
Unreleased
~~~~~~~~~~
+
+6.2.0 - 2022-11-16
+------------------
+* Adds support for LTI 1.3 Proctoring Service specification in-browser proctoring launch.
+
+ * Adds an Lti1p3ProctoringLaunchData data class. It should be included as an attribute of the Lti1p3LaunchData
+ data class to provide necessary proctoring data for a proctoring launch.
+ * Adds an LtiProctoringConsumer class. This class is used to generate LTI proctoring launch requests and to decode
+ and validate the JWT send back by the Tool with the LtiStartAssessment message.
+ * Adds an lti_1p3_proctoring_enabled BooleanField to the LtiConfiguration model. This field controls whether
+ proctoring is enabled for a particular LTI integration.
+ * Modifies the launch_gate_endpoint to support LtiStartProctoring and LtiEndAssessment LTI launch messages.
+ * Adds an start_proctoring_assessment_endpoint to support LtiStartAssessment messages from the Tool.
+ * Adds an LTI_1P3_PROCTORING_ASSESSMENT_STARTED signal. This signal is emitted when the LtiStartAssessment message is
+ sent from the Tool to inform users of the library that the LtiStartAssessment message has been received.
+
6.1.0 - 2022-11-08
------------------
* 6.0.0 broke studio functionality because it leaned more heavily on the xblock load which only worked in the LMS.
* Fix by greatly limiting when we attempt a full xblock load and bind
-
-
6.0.0 - 2022-10-24
------------------
BREAKING CHANGE:
diff --git a/docs/proctoring.rst b/docs/proctoring.rst
new file mode 100644
index 0000000000000000000000000000000000000000..20e2793a952b2314b1f0dd2ca24a346539823512
--- /dev/null
+++ b/docs/proctoring.rst
@@ -0,0 +1,95 @@
+#######################
+LTI Proctoring Features
+#######################
+
+Using LTI Proctoring Features
+*****************************
+
+Currently, this library supports a subset of the functionality in the `1EdTech Proctoring Services Specification
+<http://www.imsglobal.org/spec/proctoring/v1p0>`_. It supports the Proctoring Assessment Messages (i.e. the in-browser
+LTI proctoring launches), but it does not support the Assessment Control Service (i.e. the proctoring service calls).
+These proctoring features are currently only supported for LTI integrations using the ``CONFIG_ON_DB`` ``config_store``
+option.
+
+To enable LTI Proctoring features, you need to set the **Enable LTI Proctoring Services** field of the
+``LtiConfiguration`` model to ``True``.
+
+To start an LTI 1.3 launch with the ``LtiStartProctoring`` or ``LtiEndAssessment`` LTI message, you need to call
+the ``get_lti_1p3_launch_start_url`` Python API function with the appropriate arguments. You will need to make a request against the URL
+returned by this function. The ``launch_data`` argument will contain all data necessary for the LTI 1.3 launch. The
+``launch_data`` argument must be an instance of the ``Lti1p3LaunchData`` data class. In order to support the proctoring
+features, you must also supply a value for the ``proctoring_launch_data`` field of the ``Lti1p3LaunchData`` class. The
+``proctoring_launch_data`` argument must be an instance of the ``Lti1p3ProctoringLaunchData`` class.
+
+LTI Start Proctoring Launch
+^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Below is an example instantiation of the ``Lti1p3LaunchData`` class for an ``LtiStartProctoring`` LTI message. Please
+see the docstring of both data classes for more detailed documentation.
+
+This library implements the view that handles the ``LtiStartAssessment`` LTI message that is sent by the tool. The name
+of the URL for this view is ``lti_consumer.start_proctoring_assessment_endpoint``. You should use this URL as the
+``start_assessment_url``.
+
+.. code:: python
+
+ proctoring_start_assessment_url = urljoin(
+ <URL_ROOT>,
+ reverse('lti_consumer:lti_consumer.start_proctoring_assessment_endpoint')
+ )
+
+ proctoring_launch_data = Lti1p3ProctoringLaunchData(
+ attempt_number=<attempt_number>,
+ start_assessment_url=proctoring_start_assessment_url,
+ )
+
+ launch_data = Lti1p3LaunchData(
+ user_id=<user_id>,
+ user_role=<user_role>,
+ config_id=<config_id>,
+ resource_link_id=<resource_link_id>,
+ message_type="LtiStartProctoring",
+ proctoring_launch_data=proctoring_launch_data,
+ )
+
+ return redirect(get_lti_1p3_launch_start_url(launch_data))
+
+
+Note that for an ``LtiStartProctoring`` LTI launch message, the ``message_type`` field of the ``Lti1p3LaunchData``
+instance must be ``LtiStartProctoring`` and the ``start_assessment_url`` field of the ``Lti1p3ProctoringLaunchData``
+instance must be supplied.
+
+LTI End Assessment Launch
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Below is an example instantiation of the ``Lti1p3LaunchData`` class for an ``LtiStartProctoring`` LTI message. Please
+see the docstring of both data classes for more detailed documentation.
+
+In order to determine whether the platform should send an ``LtiEndAssessment`` LTI message to the tool, you should use
+the ``get_end_assessment_return`` Python API function. This will return a boolean representing whether the tool
+requested that the platform send an ``LtiEndAssessment`` LTI message at the end of the proctored assessment.
+
+.. code:: python
+
+ end_assessment_return = get_end_assessment_return(<user_id>, <resource_link_id>)
+
+ if end_assessment_return:
+ proctoring_launch_data = Lti1p3ProctoringLaunchData(
+ attempt_number=<attempt_number>,
+ )
+
+ launch_data = Lti1p3LaunchData(
+ user_id=<user_id>,
+ user_role=<user_role>,
+ config_id=<config_id>,
+ resource_link_id=<resource_link_id>,
+ message_type="LtiEndAssessment",
+ proctoring_launch_data=proctoring_launch_data,
+ )
+
+ return redirect(get_lti_1p3_launch_start_url(launch_data))
+
+
+Note that for an ``LtiEndAssessment`` LTI launch message, the ``message_type`` field of the ``Lti1p3LaunchData``
+instance must be ``LtiEndAssessment``. Unlike the ``LtiStartProctoring`` message, the ``start_assessment_url`` field of
+the ``Lti1p3ProctoringLaunchData`` instance should not be supplied.
diff --git a/lti_consumer/api.py b/lti_consumer/api.py
index f0f3dd7059dc72d8a392f07c68c0a96a26d19cfa..34290d574fac720c22bfd9c30fb0b8ac919dd9f4 100644
--- a/lti_consumer/api.py
+++ b/lti_consumer/api.py
@@ -12,6 +12,8 @@ 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 .utils import (
+ get_cache_key,
+ get_data_from_cache,
get_lti_1p3_context_types_claim,
get_lti_deeplinking_content_url,
get_lms_lti_keyset_link,
@@ -261,7 +263,7 @@ def validate_lti_1p3_launch_data(launch_data):
"The context_id attribute is required in the launch data if any optional context properties are provided."
)
- if launch_data.user_role not in LTI_1P3_ROLE_MAP:
+ if launch_data.user_role not in LTI_1P3_ROLE_MAP and launch_data.user_role is not None:
validation_messages.append(f"The user_role attribute {launch_data.user_role} is not a valid user_role.")
context_type = launch_data.context_type
@@ -273,7 +275,42 @@ def validate_lti_1p3_launch_data(launch_data):
f"The context_type attribute {context_type} in the launch data is not a valid context_type."
)
+ proctoring_launch_data = launch_data.proctoring_launch_data
+ if (launch_data.message_type in ["LtiStartProctoring", "LtiEndAssessment"] and not
+ proctoring_launch_data):
+ validation_messages.append(
+ "The proctoring_launch_data attribute is required if the message_type attribute is \"LtiStartProctoring\" "
+ "or \"LtiEndAssessment\"."
+ )
+
+ if (proctoring_launch_data and launch_data.message_type == "LtiStartProctoring" and not
+ proctoring_launch_data.start_assessment_url):
+ validation_messages.append(
+ "The proctoring_start_assessment_url attribute is required if the message_type attribute is "
+ "\"LtiStartProctoring\"."
+ )
+
if validation_messages:
return False, validation_messages
else:
return True, []
+
+
+def get_end_assessment_return(user_id, resource_link_id):
+ """
+ Returns the end_assessment_return value stored in the cache. This can be used by applications to determine whether
+ to invoke an LtiEndAssessment LTI launch.
+
+ Arguments:
+ * user_id: the database of the requesting User model instance
+ * resource_link_id: the resource_link_id of the LTI link for the assessment
+ """
+ end_assessment_return_key = get_cache_key(
+ app="lti",
+ key="end_assessment_return",
+ user_id=user_id,
+ resource_link_id=resource_link_id,
+ )
+ cached_end_assessment_return = get_data_from_cache(end_assessment_return_key)
+
+ return cached_end_assessment_return
diff --git a/lti_consumer/apps.py b/lti_consumer/apps.py
index e50a43198a6e757e6f544f9becc0bb0633d396f9..f3f599687bc2bbff3df6db8473e762ecfdc4dc28 100644
--- a/lti_consumer/apps.py
+++ b/lti_consumer/apps.py
@@ -26,4 +26,4 @@ class LTIConsumerApp(AppConfig):
def ready(self):
# pylint: disable=unused-import,import-outside-toplevel
- import lti_consumer.signals
+ from lti_consumer.signals import signals
diff --git a/lti_consumer/data.py b/lti_consumer/data.py
index 7d2bfcfe7c1a7fa42d5323d0660042ba3d8d0000..b450115db5d3becfaed4a1cb8bc41357ed5e8c43 100644
--- a/lti_consumer/data.py
+++ b/lti_consumer/data.py
@@ -3,7 +3,30 @@ This modules provides public data structures to represent LTI 1.3 launch data th
by users of this library.
"""
-from attrs import define, field
+from attrs import define, field, validators
+
+
+@define
+class Lti1p3ProctoringLaunchData:
+ """
+ The Lti1p3ProctoringLaunchData class contains data necessary and related to an LTI 1.3 proctoring launch.
+ It is a mechanism to share launch data between apps. Applications using this library should use the
+ Lti1p3ProctoringLaunchData class to supply contextually defined or stored launch data to the LTI 1.3 proctoring
+ launch.
+
+ Lti1p3ProctoringLaunchData is intended to be initialized and included as an attribute of an instance of the
+ Lti1p3LaunchData class.
+
+ * attempt_number (required): The number of the user's attempt in the assessment. The attempt number
+ should be an integer that starts with 1 and that is incremented by 1 for each of user's subsequent attempts at
+ the assessment.
+ * start_assessment_url (conditionally required): The Platform URL that the Tool will send the start
+ assessment message to after it has completed the proctoring setup and verification. This attribute is required
+ if the message_type attribute of the Lti1p3LaunchData instance is "LtiStartProctoring". It is optional and
+ unused otherwise.
+ """
+ attempt_number = field()
+ start_assessment_url = field(default=None)
@define
@@ -13,34 +36,53 @@ class Lti1p3LaunchData:
launch data between apps. Applications using this library should use the Lti1p3LaunchData class to supply
contextually defined or stored launch data to the generic LTI 1.3 launch.
- * user_id (required): the user's unique identifier
- * user_role (required): the user's role as one of the keys in LTI_1P3_ROLE_MAP: staff, instructor, student, or
- guest
- * config_id (required): the config_id field of an LtiConfiguration to use for the launch
- * resource_link_id (required): a unique identifier that is guaranteed to be unique for each placement of the LTI
- link
- * launch_presentation_document_target (optional): the document_target property of the launch_presentation claim; it
- describes the kind of browser window or frame from which the user launched inside the message sender's system;
- it is one of frame, iframe, or window; it defaults to iframe
- * message_type (optional): the message type of the eventual LTI launch; defaults to LtiResourceLinkRequest
- * context_id (conditionally required): the id property of the context claim; the stable, unique identifier for the
- context; if any of the context properties are provided, context_id is required
- * context_type (optional): the type property of the context claim; a list of some combination of the following valid
- context_types: group, course_offering, course_section, or course_template
- * context_title (optional): the title proerty of the context claim; a short, descriptive name for the context
- * context_label (optional): the label property of the context claim; a full, descriptive name for the context
- * deep_linking_context_item_id (optional): the database id of the LtiDlContentItem that should be used for the LTI
- 1.3 Deep Linking launch; this is used when the LTI 1.3 Deep Linking launch is a regular LTI resource link
- request using a content item that was configured via a previous LTI 1.3 Deep Linking request
+ * user_id (required): A unique identifier for the user that is requesting the LTI 1.3 launch. If the optional
+ attribute external_user_id is provided, user_id will only be used internally and will not be shared externally.
+ If external_user_id is not provided, user_id will be shared externally, and then it must be stable to the
+ issuer.
+ * user_role (required): The user's role as one of the keys in LTI_1P3_ROLE_MAP: staff, instructor, student, or
+ guest. It can also be None if the empty list should be sent in the LTI launch message.
+ * config_id (required): The config_id field of an LtiConfiguration to use for the launch.
+ * resource_link_id (required): A unique identifier that is guaranteed to be unique for each placement of the LTI
+ link.
+ * external_user_id (optional): A unique identifier for the user that is requesting the LTI 1.3 launch that can be
+ shared externally. The identifier must be stable to the issuer. This value will be sent to the the Tool in the
+ form of both the login_hint in the login initiation request and the sub claim in the ID token of the LTI 1.3
+ launch message. Use this attribute to specify what value to send as this claim. Otherwise, the value of the
+ required user_id attribute will be used.
+ * launch_presentation_document_target (optional): The document_target property of the launch_presentation claim. It
+ describes the kind of browser window or frame from which the user launched inside the message sender's system.
+ It is one of frame, iframe, or window; it defaults to iframe.
+ * launch_presentation_return_url (optional): A URL where the Tool can redirect the learner after the learner
+ completes the activity or is unable to start the activity.
+ * message_type (optional): The message type of the eventual LTI launch. It defaults to LtiResourceLinkRequest.
+ * context_id (conditionally required): The id property of the context claim. It is the stable, unique identifier for
+ the context. It is required if any of the context properties are provided.
+ * context_type (optional): The type property of the context claim. It is a list of some combination of the following
+ valid context_types: group, course_offering, course_section, or course_template.
+ * context_title (optional): The title proerty of the context claim. It is a short, descriptive name for the context.
+ * context_label (optional): The label property of the context claim. It is a full, descriptive name for the context.
+ * deep_linking_context_item_id (optional): The database id of the LtiDlContentItem that should be used for the LTI
+ 1.3 Deep Linking launch. It is used when the LTI 1.3 Deep Linking launch is a regular LTI resource link
+ request using a content item that was configured via a previous LTI 1.3 Deep Linking request.
+ * 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".
"""
user_id = field()
user_role = field()
config_id = field()
resource_link_id = field()
- launch_presentation_document_target = field(default="iframe")
+ external_user_id = field(default=None)
+ launch_presentation_document_target = field(default=None)
+ launch_presentation_return_url = field(default=None)
message_type = field(default="LtiResourceLinkRequest")
context_id = field(default=None)
context_type = field(default=None)
context_title = field(default=None)
context_label = field(default=None)
deep_linking_content_item_id = field(default=None)
+ proctoring_launch_data = field(
+ default=None,
+ validator=validators.optional((validators.instance_of(Lti1p3ProctoringLaunchData))),
+ )
diff --git a/lti_consumer/lti_1p3/constants.py b/lti_consumer/lti_1p3/constants.py
index 5f85d2974820e8fb112208477555d40b423a1a9c..d00e3c203cdb29f5256b6f40da555b02a01ca6c1 100644
--- a/lti_consumer/lti_1p3/constants.py
+++ b/lti_consumer/lti_1p3/constants.py
@@ -83,3 +83,6 @@ class LTI_1P3_CONTEXT_TYPE(Enum): # pylint: disable=invalid-name
course_offering = 'http://purl.imsglobal.org/vocab/lis/v2/course#CourseOffering'
course_section = 'http://purl.imsglobal.org/vocab/lis/v2/course#CourseSection'
course_template = 'http://purl.imsglobal.org/vocab/lis/v2/course#CourseTemplate'
+
+
+LTI_PROCTORING_DATA_KEYS = ['attempt_number', 'resource_link_id', 'session_data', 'start_assessment_url']
diff --git a/lti_consumer/lti_1p3/consumer.py b/lti_consumer/lti_1p3/consumer.py
index c45a608384a58e0a758727776f807193e8557f03..6360227de84f07c2f2340fcf784ef9cb9a606ef5 100644
--- a/lti_consumer/lti_1p3/consumer.py
+++ b/lti_consumer/lti_1p3/consumer.py
@@ -4,7 +4,8 @@ LTI 1.3 Consumer implementation
import logging
from urllib.parse import urlencode
-from lti_consumer.utils import cache_lti_1p3_launch_data, get_data_from_cache
+from lti_consumer.lti_1p3.exceptions import InvalidClaimValue
+from lti_consumer.utils import cache_lti_1p3_launch_data, check_token_claim, get_data_from_cache
from . import constants, exceptions
from .constants import (
@@ -13,6 +14,7 @@ from .constants import (
LTI_1P3_ACCESS_TOKEN_REQUIRED_CLAIMS,
LTI_1P3_ACCESS_TOKEN_SCOPES,
LTI_1P3_CONTEXT_TYPE,
+ LTI_PROCTORING_DATA_KEYS,
)
from .key_handlers import ToolKeyHandler, PlatformKeyHandler
from .ags import LtiAgs
@@ -101,7 +103,7 @@ class LtiConsumer1p3:
"""
Generates OIDC url with parameters
"""
- user_id = launch_data.user_id
+ user_id = launch_data.external_user_id if launch_data.external_user_id else launch_data.user_id
# Set the launch_data in the cache. An LTI 1.3 launch involves two "legs" - the third party initiated
# login request (the preflight request) and the actual launch -, and this information must be shared between
@@ -193,25 +195,27 @@ class LtiConsumer1p3:
def set_launch_presentation_claim(
self,
- document_target="iframe"
+ document_target=None,
+ return_url=None,
):
"""
Optional: Set launch presentation claims
http://www.imsglobal.org/spec/lti/v1p3/#launch-presentation-claim
"""
- if document_target not in ['iframe', 'frame', 'window']:
+ if document_target is not None and document_target not in ['iframe', 'frame', 'window']:
raise ValueError("Invalid launch presentation format.")
+ lti_claim_launch_presentation = {}
+
+ if document_target:
+ lti_claim_launch_presentation.update({"document_target": document_target})
+
+ if return_url:
+ lti_claim_launch_presentation.update({"return_url": return_url})
+
self.lti_claim_launch_presentation = {
- # Launch presentation claim
- "https://purl.imsglobal.org/spec/lti/claim/launch_presentation": {
- # Can be one of: iframe, frame, window
- "document_target": document_target,
- # TODO: Add support for `return_url` handler to allow the tool
- # to return error messages back to the lms.
- # See the spec referenced above for more information.
- },
+ "https://purl.imsglobal.org/spec/lti/claim/launch_presentation": lti_claim_launch_presentation,
}
def set_context_claim(
@@ -704,3 +708,193 @@ class LtiAdvantageConsumer(LtiConsumer1p3):
# Include LTI NRPS claim inside the LTI Launch message
self.set_extra_claim(self.nrps.get_lti_nrps_launch_claim())
+
+
+class LtiProctoringConsumer(LtiConsumer1p3):
+ """
+ This class is an LTI Proctoring Services LTI consumer implementation.
+
+ It builds on top of the LtiConsumer1p3 and adds support for the LTI Proctoring Services specification. The
+ specification can be found here: http://www.imsglobal.org/spec/proctoring/v1p0.
+
+ This consumer currently only supports the "Assessment Proctoring Messages" and the proctoring assessment flow.
+ It does not currently support the Assessment Control Service.
+
+ The LtiProctoringConsumer requires necessary context to work properly, including data like attempt_number,
+ resource_link, etc. This information is provided to the consumer through the set_proctoring_data method, which
+ is called from the consuming context to pass in necessary data.
+ """
+ def __init__(
+ self,
+ iss,
+ lti_oidc_url,
+ lti_launch_url,
+ client_id,
+ deployment_id,
+ rsa_key,
+ rsa_key_id,
+ tool_key=None,
+ tool_keyset_url=None,
+ ):
+ """
+ Initialize the LtiProctoringConsumer by delegating to LtiConsumer1p3's __init__ method.
+ """
+ super().__init__(
+ iss,
+ lti_oidc_url,
+ lti_launch_url,
+ client_id,
+ deployment_id,
+ rsa_key,
+ rsa_key_id,
+ tool_key,
+ tool_keyset_url
+ )
+ self.proctoring_data = {}
+
+ def set_proctoring_data(self, **kwargs):
+ """
+ Sets the self.proctoring_data dictionary with the provided kwargs, so long as a given key is in
+ LTI_PROCTORING_DATA_KEYS.
+ """
+ for key, value in kwargs.items():
+ if key in LTI_PROCTORING_DATA_KEYS:
+ self.proctoring_data[key] = value
+
+ def _get_base_claims(self):
+ """
+ Returns claims common to all LTI Proctoring Services LTI launch messages, to be used when creating LTI launch
+ messages.
+ """
+ proctoring_claims = {
+ "https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number": self.proctoring_data.get("attempt_number"),
+ "https://purl.imsglobal.org/spec/lti-ap/claim/session_data": self.proctoring_data.get("session_data"),
+ }
+
+ return proctoring_claims
+
+ def get_start_proctoring_claims(self):
+ """
+ Returns claims specific to LTI Proctoring Services LtiStartProctoring LTI launch message,
+ to be injected into the LTI launch message.
+ """
+ proctoring_claims = self._get_base_claims()
+ proctoring_claims.update({
+ "https://purl.imsglobal.org/spec/lti/claim/message_type": "LtiStartProctoring",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/start_assessment_url":
+ self.proctoring_data.get("start_assessment_url"),
+ })
+
+ return proctoring_claims
+
+ def get_end_assessment_claims(self):
+ """
+ Returns claims specific to LTI Proctoring Services LtiEndAssessment LTI launch message,
+ to be injected into the LTI launch message.
+ """
+ proctoring_claims = self._get_base_claims()
+ proctoring_claims.update({
+ "https://purl.imsglobal.org/spec/lti/claim/message_type": "LtiEndAssessment",
+ })
+
+ return proctoring_claims
+
+ def generate_launch_request(
+ self,
+ preflight_response,
+ ):
+ """
+ Builds and return LTI launch message for proctoring.
+
+ Overrides LtiConsumer1p3's method to include proctoring specific launch claims. Leverages
+ the set_extra_claim method to include these additional claims in the LTI launch message.
+ """
+ lti_message_hint = preflight_response.get('lti_message_hint')
+ launch_data = get_data_from_cache(lti_message_hint)
+
+ if launch_data.message_type == "LtiStartProctoring":
+ proctoring_claims = self.get_start_proctoring_claims()
+ elif launch_data.message_type == "LtiEndAssessment":
+ proctoring_claims = self.get_end_assessment_claims()
+ else:
+ raise ValueError('lti_message_hint must \"LtiStartProctoring\" or \"LtiEndAssessment\".')
+
+ self.set_extra_claim(proctoring_claims)
+
+ return super().generate_launch_request(preflight_response)
+
+ def check_and_decode_token(self, token):
+ """
+ Decodes a Tool JWT token and validates OAuth and LTI Proctoring Services specificatin related claims. Returns a
+ dictionary representation of key proctoring claims in the Tool JWT token.
+
+ Arguments:
+ * token (string): a JWT
+ """
+ # Decode token and check expiration.
+ proctoring_response = self.tool_jwt.validate_and_decode(token)
+
+ # -------------------------
+ # Check Required LTI Claims
+ # -------------------------
+
+ # Check that the response message_type claim is "LtiStartAssessment".
+ claim_key = "https://purl.imsglobal.org/spec/lti/claim/message_type"
+ check_token_claim(
+ proctoring_response,
+ claim_key,
+ "LtiStartAssessment",
+ f"Token's {claim_key} claim should be LtiStartAssessment."
+ )
+
+ # # Check that the response version claim is "1.3.0".
+ claim_key = "https://purl.imsglobal.org/spec/lti/claim/version"
+ check_token_claim(
+ proctoring_response,
+ claim_key,
+ "1.3.0",
+ f"Token's {claim_key} claim should be 1.3.0."
+ )
+
+ # Check that the response session_data claim is the correct anti-CSRF token.
+ claim_key = "https://purl.imsglobal.org/spec/lti-ap/claim/session_data"
+ check_token_claim(
+ proctoring_response,
+ claim_key,
+ self.proctoring_data.get("session_data"),
+ f"Token's {claim_key} claim is not correct."
+ )
+
+ # TODO: This is a special case. Right now, the library doesn't support additional claims within the
+ # resource_link claim. Once it does, we should check the entire claim instead of just the id. For now, check
+ # that the resource_link claim is supplied and that the id attribute is correct.
+ claim_key = "https://purl.imsglobal.org/spec/lti/claim/resource_link"
+ resource_link = proctoring_response.get(claim_key)
+ check_token_claim(
+ proctoring_response,
+ claim_key,
+ )
+
+ resource_link_id = resource_link.get("id")
+ if self.proctoring_data.get("resource_link_id") != resource_link_id:
+ raise InvalidClaimValue(f"Token's {claim_key} claim is not correct.")
+
+ claim_key = "https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number"
+ check_token_claim(
+ proctoring_response,
+ claim_key,
+ self.proctoring_data.get("attempt_number"),
+ f"Token's {claim_key} claim is not correct."
+ )
+
+ response = {
+ 'end_assessment_return': proctoring_response.get(
+ "https://purl.imsglobal.org/spec/lti-ap/claim/end_assessment_return",
+ ),
+ 'verified_user': proctoring_response.get("https://purl.imsglobal.org/spec/lti-ap/claim/verified_user", {}),
+ 'resource_link': proctoring_response["https://purl.imsglobal.org/spec/lti/claim/resource_link"],
+ 'session_data': proctoring_response["https://purl.imsglobal.org/spec/lti-ap/claim/session_data"],
+ 'attempt_number': proctoring_response["https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number"],
+ }
+
+ return response
diff --git a/lti_consumer/lti_1p3/exceptions.py b/lti_consumer/lti_1p3/exceptions.py
index 7ef835b906b4346c79e10257f447ac1e9867d9d9..0b2c67c62987217ead6f9710d8e2dbe20ca70b92 100644
--- a/lti_consumer/lti_1p3/exceptions.py
+++ b/lti_consumer/lti_1p3/exceptions.py
@@ -30,6 +30,10 @@ class NoSuitableKeys(Lti1p3Exception):
message = "JWKS could not be loaded from the URL."
+class BadJwtSignature(Lti1p3Exception):
+ message = "The JWT signature is invalid."
+
+
class UnknownClientId(Lti1p3Exception):
pass
diff --git a/lti_consumer/lti_1p3/key_handlers.py b/lti_consumer/lti_1p3/key_handlers.py
index 7f52b4a70d0d43d1ce0d3e8bb924853fe662a757..64048c5df05910f1a60f2ed24dba8f66a66e669c 100644
--- a/lti_consumer/lti_1p3/key_handlers.py
+++ b/lti_consumer/lti_1p3/key_handlers.py
@@ -10,7 +10,7 @@ import time
import json
from Cryptodome.PublicKey import RSA
-from jwkest import BadSyntax, WrongNumberOfParts, jwk
+from jwkest import BadSignature, BadSyntax, WrongNumberOfParts, jwk
from jwkest.jwk import RSAKey, load_jwks_from_url
from jwkest.jws import JWS, NoSuitableSigningKeys
from jwkest.jwt import JWT
@@ -124,6 +124,8 @@ class ToolKeyHandler:
raise exceptions.NoSuitableKeys() from err
except (BadSyntax, WrongNumberOfParts) as err:
raise exceptions.MalformedJwtToken() from err
+ except BadSignature as err:
+ raise exceptions.BadJwtSignature() from err
class PlatformKeyHandler:
diff --git a/lti_consumer/lti_1p3/tests/test_consumer.py b/lti_consumer/lti_1p3/tests/test_consumer.py
index 423c414e667b55cd84f57be733688ae84b686f66..9528e344c2a927cf43d5924b98b205de25c526aa 100644
--- a/lti_consumer/lti_1p3/tests/test_consumer.py
+++ b/lti_consumer/lti_1p3/tests/test_consumer.py
@@ -18,8 +18,10 @@ from lti_consumer.lti_1p3 import exceptions
from lti_consumer.lti_1p3.ags import LtiAgs
from lti_consumer.lti_1p3.deep_linking import LtiDeepLinking
from lti_consumer.lti_1p3.nprs import LtiNrps
-from lti_consumer.lti_1p3.constants import LTI_1P3_CONTEXT_TYPE
-from lti_consumer.lti_1p3.consumer import LtiAdvantageConsumer, LtiConsumer1p3
+from lti_consumer.lti_1p3.constants import LTI_1P3_CONTEXT_TYPE, LTI_PROCTORING_DATA_KEYS
+from lti_consumer.lti_1p3.consumer import LtiAdvantageConsumer, LtiConsumer1p3, LtiProctoringConsumer
+from lti_consumer.lti_1p3.exceptions import InvalidClaimValue, MissingRequiredClaim
+
# Variables required for testing and verification
ISS = "http://test-platform.example/"
@@ -247,12 +249,18 @@ class TestLti1p3Consumer(TestCase):
Check if setting presentation claim data works
"""
self._setup_lti_launch_data()
- self.lti_consumer.set_launch_presentation_claim(document_target=target)
+
+ return_url = "return_url"
+ self.lti_consumer.set_launch_presentation_claim(
+ document_target=target,
+ return_url=return_url,
+ )
self.assertEqual(
self.lti_consumer.lti_claim_launch_presentation,
{
"https://purl.imsglobal.org/spec/lti/claim/launch_presentation": {
- "document_target": target
+ "document_target": target,
+ "return_url": return_url,
}
}
)
@@ -269,7 +277,8 @@ class TestLti1p3Consumer(TestCase):
self.assertEqual(
decoded["https://purl.imsglobal.org/spec/lti/claim/launch_presentation"],
{
- "document_target": target
+ "document_target": target,
+ "return_url": return_url,
}
)
@@ -846,3 +855,260 @@ class TestLtiAdvantageConsumer(TestCase):
}
}
)
+
+
+@ddt.ddt
+class TestLtiProctoringConsumer(TestCase):
+ """
+ Unit tests for LtiProctoringConsumer
+ """
+ def setUp(self):
+ super().setUp()
+
+ # Set up consumer
+ self.lti_consumer = LtiProctoringConsumer(
+ iss=ISS,
+ lti_oidc_url=OIDC_URL,
+ lti_launch_url=LAUNCH_URL,
+ client_id=CLIENT_ID,
+ deployment_id=DEPLOYMENT_ID,
+ rsa_key=RSA_KEY,
+ rsa_key_id=RSA_KEY_ID,
+ # Use the same key for testing purposes
+ tool_key=RSA_KEY
+ )
+
+ self.preflight_response = {}
+
+ def _setup_proctoring(self):
+ """
+ Sets up data necessary for a proctoring LTI launch.
+ """
+ # Set LTI Consumer parameters
+ self.preflight_response = {
+ "client_id": CLIENT_ID,
+ "redirect_uri": LAUNCH_URL,
+ "nonce": NONCE,
+ "state": STATE,
+ "lti_message_hint": "lti_message_hint",
+ }
+ self.lti_consumer.set_user_data("1", "student")
+ self.lti_consumer.set_resource_link_claim("resource_link_id")
+
+ def get_launch_data(self, **kwargs):
+ """
+ Returns a sample instance of Lti1p3LaunchData.
+ """
+ launch_data_kwargs = {
+ "user_id": "user_id",
+ "user_role": "student",
+ "config_id": "1",
+ "resource_link_id": "resource_link_id",
+ }
+
+ launch_data_kwargs.update(kwargs)
+
+ return Lti1p3LaunchData(**launch_data_kwargs)
+
+ @ddt.data(*LTI_PROCTORING_DATA_KEYS)
+ def test_valid_set_proctoring_data(self, key):
+ """
+ Ensures that valid proctoring data can be set on an instance of LtiProctoringConsumer.
+ """
+ value = "test_value"
+ self.lti_consumer.set_proctoring_data(**{key: value})
+
+ actual_value = self.lti_consumer.proctoring_data[key]
+
+ self.assertEqual(value, actual_value)
+
+ def test_invalid_set_proctoring_data(self):
+ """
+ Ensures that an attempt to set invalid proctoring data on an instance of LtiProctoringConsumer does not update
+ the consumer's proctoring_data.
+ """
+ self.lti_consumer.set_proctoring_data(test_key="test_value")
+
+ self.assertEqual({}, self.lti_consumer.proctoring_data)
+
+ def test_get_start_proctoring_claims(self):
+ """
+ Ensures that the correct claims are returned for a LtiStartProctoring LTI launch message.
+ """
+ self.lti_consumer.set_proctoring_data(
+ attempt_number="attempt_number",
+ session_data="session_data",
+ start_assessment_url="start_assessment_url",
+ )
+
+ actual_start_proctoring_claims = self.lti_consumer.get_start_proctoring_claims()
+
+ expected_start_proctoring_claims = {
+ "https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number": "attempt_number",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/session_data": "session_data",
+ "https://purl.imsglobal.org/spec/lti/claim/message_type": "LtiStartProctoring",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/start_assessment_url": "start_assessment_url",
+ }
+
+ self.assertEqual(expected_start_proctoring_claims, actual_start_proctoring_claims)
+
+ def test_get_end_assessment_claims(self):
+ """
+ Ensures that the correct claims are returned for a LtiEndAssessment LTI launch message.
+ """
+ self.lti_consumer.set_proctoring_data(
+ attempt_number="attempt_number",
+ session_data="session_data",
+ )
+
+ actual_get_end_assessment_claims = self.lti_consumer.get_end_assessment_claims()
+
+ expected_get_end_assessment_claims = {
+ "https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number": "attempt_number",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/session_data": "session_data",
+ "https://purl.imsglobal.org/spec/lti/claim/message_type": "LtiEndAssessment",
+ }
+
+ self.assertEqual(expected_get_end_assessment_claims, actual_get_end_assessment_claims)
+
+ @ddt.data("LtiStartProctoring", "LtiEndAssessment")
+ @patch('lti_consumer.lti_1p3.consumer.get_data_from_cache')
+ def test_generate_launch_request(self, message_type, mock_get_data_from_cache):
+ """
+ Ensures that the correct claims are included in LTI launch messages for the LtiStartProctoring and
+ LtiEndAssessment launch message types.
+ """
+
+ mock_launch_data = self.get_launch_data(message_type=message_type)
+ mock_get_data_from_cache.return_value = mock_launch_data
+
+ self._setup_proctoring()
+
+ self.lti_consumer.set_proctoring_data(
+ attempt_number="attempt_number",
+ session_data="session_data",
+ start_assessment_url="start_assessment_url",
+ )
+
+ token = self.lti_consumer.generate_launch_request(
+ self.preflight_response,
+ )['id_token']
+
+ decoded_token = self.lti_consumer.key_handler.validate_and_decode(token)
+
+ expected_claims = {}
+
+ if message_type == "LtiStartProctoring":
+ expected_claims = self.lti_consumer.get_start_proctoring_claims()
+ else:
+ expected_claims = self.lti_consumer.get_end_assessment_claims()
+
+ decoded_token_claims = decoded_token.items()
+ for claim in expected_claims.items():
+ self.assertIn(claim, decoded_token_claims)
+
+ @patch('lti_consumer.lti_1p3.consumer.get_data_from_cache')
+ def test_generate_launch_request_invalid_message(self, mock_get_data_from_cache):
+ """
+ Ensures that a ValueError is raised if the launch_data.message_type is not LtiStartProctoring or
+ LtiEndAssessment.
+ """
+
+ mock_launch_data = self.get_launch_data(message_type="LtiResourceLinkRequest")
+ mock_get_data_from_cache.return_value = mock_launch_data
+
+ self._setup_proctoring()
+
+ self.lti_consumer.set_proctoring_data(
+ attempt_number="attempt_number",
+ session_data="session_data",
+ start_assessment_url="start_assessment_url",
+ )
+
+ with self.assertRaises(ValueError):
+ _ = self.lti_consumer.generate_launch_request(
+ self.preflight_response,
+ )['id_token']
+
+ @ddt.data(
+ "https://purl.imsglobal.org/spec/lti/claim/message_type",
+ "https://purl.imsglobal.org/spec/lti/claim/version",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/session_data",
+ "https://purl.imsglobal.org/spec/lti/claim/resource_link",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number",
+ )
+ def test_invalid_check_and_decode_token(self, claim_key):
+ """
+ Ensures that LtiStartAssessment JWTs are correctly validated; ensures that missing or incorrect claims cause an
+ InvalidClaimValue or MissingRequiredClaim exception to be raised.
+ """
+ self.lti_consumer.set_proctoring_data(
+ attempt_number="attempt_number",
+ session_data="session_data",
+ start_assessment_url="start_assessment_url",
+ resource_link_id="resource_link_id",
+ )
+
+ start_assessment_response = {
+ "https://purl.imsglobal.org/spec/lti/claim/message_type": "LtiStartAssessment",
+ "https://purl.imsglobal.org/spec/lti/claim/version": "1.3.0",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/session_data": "session_data",
+ "https://purl.imsglobal.org/spec/lti/claim/resource_link": {"id": "resource_link_id"},
+ "https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number": "attempt_number",
+ }
+
+ # Check invalid claim values.
+ start_assessment_response[claim_key] = {}
+ encoded_token = self.lti_consumer.key_handler.encode_and_sign(
+ message=start_assessment_response,
+ expiration=3600
+ )
+
+ with self.assertRaises(InvalidClaimValue):
+ self.lti_consumer.check_and_decode_token(encoded_token)
+
+ # Check missing claims.
+ del start_assessment_response[claim_key]
+ encoded_token = self.lti_consumer.key_handler.encode_and_sign(
+ message=start_assessment_response,
+ expiration=3600
+ )
+
+ with self.assertRaises(MissingRequiredClaim):
+ self.lti_consumer.check_and_decode_token(encoded_token)
+
+ def test_valid_check_and_decode_token(self):
+ """
+ Ensures that a valid LtiStartAssessment JWT is validated successfully.
+ """
+ self.lti_consumer.set_proctoring_data(
+ attempt_number="attempt_number",
+ session_data="session_data",
+ start_assessment_url="start_assessment_url",
+ resource_link_id="resource_link_id",
+ )
+
+ start_assessment_response = {
+ "https://purl.imsglobal.org/spec/lti/claim/message_type": "LtiStartAssessment",
+ "https://purl.imsglobal.org/spec/lti/claim/version": "1.3.0",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/session_data": "session_data",
+ "https://purl.imsglobal.org/spec/lti/claim/resource_link": {"id": "resource_link_id"},
+ "https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number": "attempt_number",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/verified_user": {"name": "Bob"},
+ "https://purl.imsglobal.org/spec/lti-ap/claim/end_assessment_return": "end_assessment_return",
+ }
+
+ encoded_token = self.lti_consumer.key_handler.encode_and_sign(
+ message=start_assessment_response,
+ expiration=3600
+ )
+
+ response = self.lti_consumer.check_and_decode_token(encoded_token)
+ expected_response = {
+ "end_assessment_return": "end_assessment_return",
+ "verified_user": {"name": "Bob"},
+ "resource_link": {"id": "resource_link_id"},
+ "session_data": "session_data",
+ "attempt_number": "attempt_number"
+ }
+ self.assertEqual(expected_response, response)
diff --git a/lti_consumer/lti_1p3/tests/test_key_handlers.py b/lti_consumer/lti_1p3/tests/test_key_handlers.py
index ab7a7734e299f57bd9b170a5a76ad87c6525938a..e087ad8da4ce59a1949a5cce33f68e71df7cddf7 100644
--- a/lti_consumer/lti_1p3/tests/test_key_handlers.py
+++ b/lti_consumer/lti_1p3/tests/test_key_handlers.py
@@ -8,6 +8,7 @@ from unittest.mock import patch
import ddt
from Cryptodome.PublicKey import RSA
from django.test.testcases import TestCase
+from jwkest import BadSignature
from jwkest.jwk import RSAKey, load_jwks
from jwkest.jws import JWS
@@ -300,3 +301,20 @@ class TestToolKeyHandler(TestCase):
# Decode and check results
with self.assertRaises(exceptions.NoSuitableKeys):
key_handler.validate_and_decode(signed)
+
+ @patch("lti_consumer.lti_1p3.key_handlers.JWS.verify_compact")
+ def test_validate_and_decode_bad_signature(self, mock_verify_compact):
+ mock_verify_compact.side_effect = BadSignature()
+
+ key_handler = ToolKeyHandler()
+
+ message = {
+ "test": "test_message",
+ "iat": 1000,
+ "exp": 1200,
+ }
+ signed = create_jwt(self.key, message)
+
+ # Decode and check results
+ with self.assertRaises(exceptions.BadJwtSignature):
+ key_handler.validate_and_decode(signed)
diff --git a/lti_consumer/lti_xblock.py b/lti_consumer/lti_xblock.py
index 9973e4395917bec2af8bcea551442e282b3c4557..311f9fed743b51d88b816f21ff9b312e9a275fd2 100644
--- a/lti_consumer/lti_xblock.py
+++ b/lti_consumer/lti_xblock.py
@@ -785,7 +785,19 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
return '', ''
@property
- def user_id(self):
+ def lms_user_id(self):
+ """
+ Returns the edx-platform database user id for the current user.
+ """
+ user_id = self.runtime.service(self, 'user').get_current_user().opt_attrs.get(
+ 'edx-platform.user_id', None)
+
+ if user_id is None:
+ raise LtiError(self.ugettext("Could not get user id for current request"))
+ return user_id
+
+ @property
+ def anonymous_user_id(self):
"""
Returns the opaque anonymous_student_id for the current user.
This defaults to 'student' when testing in studio.
@@ -863,7 +875,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
return "{context}:{resource_link}:{user_id}".format(
context=urllib.parse.quote(self.context_id),
resource_link=self.resource_link_id,
- user_id=self.user_id
+ user_id=self.anonymous_user_id
)
@property
@@ -1100,7 +1112,7 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
# return a 400 response with an appropriate error template.
try:
real_user_data = self.extract_real_user_data()
- user_id = self.user_id
+ user_id = self.anonymous_user_id
role = self.role
# Convert the LMS role into an LTI 1.1 role.
@@ -1429,10 +1441,11 @@ class LtiConsumerXBlock(StudioEditableXBlockMixin, XBlock):
course_key = str(location.course_key)
launch_data = Lti1p3LaunchData(
- user_id=self.external_user_id,
+ user_id=self.lms_user_id,
user_role=self.role,
config_id=config_id,
resource_link_id=str(location),
+ external_user_id=self.external_user_id,
launch_presentation_document_target="iframe",
context_id=course_key,
context_type=["course_offering"],
diff --git a/lti_consumer/migrations/0016_lticonfiguration_lti_1p3_proctoring_enabled.py b/lti_consumer/migrations/0016_lticonfiguration_lti_1p3_proctoring_enabled.py
new file mode 100644
index 0000000000000000000000000000000000000000..ea218b5bce4e3fd4f318eafee295d3e73f739722
--- /dev/null
+++ b/lti_consumer/migrations/0016_lticonfiguration_lti_1p3_proctoring_enabled.py
@@ -0,0 +1,18 @@
+# Generated by Django 3.2.14 on 2022-10-13 20:26
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('lti_consumer', '0015_add_additional_1p3_fields'),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name='lticonfiguration',
+ name='lti_1p3_proctoring_enabled',
+ field=models.BooleanField(default=False, help_text='Enable LTI Proctoring Services', verbose_name='Enable LTI Proctoring Services'),
+ ),
+ ]
diff --git a/lti_consumer/models.py b/lti_consumer/models.py
index 0254f92ce6eb3fc00ed8fb5d3c14a8baf8597e0e..2eb9ae013f3fae1af34bd57e4e982b775b0e2f28 100644
--- a/lti_consumer/models.py
+++ b/lti_consumer/models.py
@@ -20,7 +20,7 @@ from lti_consumer.filters import get_external_config_from_filter
# LTI 1.1
from lti_consumer.lti_1p1.consumer import LtiConsumer1p1
# LTI 1.3
-from lti_consumer.lti_1p3.consumer import LtiAdvantageConsumer
+from lti_consumer.lti_1p3.consumer import LtiAdvantageConsumer, LtiProctoringConsumer
from lti_consumer.lti_1p3.key_handlers import PlatformKeyHandler
from lti_consumer.plugin import compat
from lti_consumer.plugin.compat import request_cached
@@ -227,6 +227,13 @@ class LtiConfiguration(models.Model):
'grades.'
)
+ # LTI Proctoring Service Related Variables
+ lti_1p3_proctoring_enabled = models.BooleanField(
+ "Enable LTI Proctoring Services",
+ default=False,
+ help_text='Enable LTI Proctoring Services',
+ )
+
def clean(self):
if self.config_store == self.CONFIG_ON_XBLOCK and self.location is None:
raise ValidationError({
@@ -245,6 +252,12 @@ class LtiConfiguration(models.Model):
"lti_1p3_tool_public_key or lti_1p3_tool_keyset_url."
),
})
+ if (self.version == self.LTI_1P3 and self.config_store in [self.CONFIG_ON_XBLOCK, self.CONFIG_EXTERNAL] and
+ self.lti_1p3_proctoring_enabled):
+ raise ValidationError({
+ "config_store": _("CONFIG_ON_XBLOCK and CONFIG_EXTERNAL are not supported for "
+ "LTI 1.3 Proctoring Services."),
+ })
try:
consumer = self.get_lti_consumer()
except NotImplementedError:
@@ -470,9 +483,18 @@ class LtiConfiguration(models.Model):
Uses the `config_store` variable to determine where to
look for the configuration and instance the class.
"""
+ consumer_class = LtiAdvantageConsumer
+ # LTI Proctoring Services is not currently supported for CONFIG_ON_XBLOCK or CONFIG_EXTERNAL.
+ # NOTE: This currently prevents an LTI Consumer from supporting both the LTI 1.3 proctoring feature and the LTI
+ # Advantage services. We plan to address this. Follow this issue:
+ # https://github.com/openedx/xblock-lti-consumer/issues/303.
+ if self.lti_1p3_proctoring_enabled and self.config_store == self.CONFIG_ON_DB:
+ consumer_class = LtiProctoringConsumer
+
if self.config_store == self.CONFIG_ON_XBLOCK:
block = compat.load_enough_xblock(self.location)
- consumer = LtiAdvantageConsumer(
+
+ consumer = consumer_class(
iss=get_lms_base(),
lti_oidc_url=block.lti_1p3_oidc_url,
lti_launch_url=block.lti_1p3_launch_url,
@@ -488,7 +510,7 @@ class LtiConfiguration(models.Model):
tool_keyset_url=block.lti_1p3_tool_keyset_url,
)
elif self.config_store == self.CONFIG_ON_DB:
- consumer = LtiAdvantageConsumer(
+ consumer = consumer_class(
iss=get_lms_base(),
lti_oidc_url=self.lti_1p3_oidc_url,
lti_launch_url=self.lti_1p3_launch_url,
@@ -508,9 +530,11 @@ class LtiConfiguration(models.Model):
# or CONFIG_ON_DB.
raise NotImplementedError
- self._setup_lti_1p3_ags(consumer)
- self._setup_lti_1p3_deep_linking(consumer)
- self._setup_lti_1p3_nrps(consumer)
+ if isinstance(consumer, LtiAdvantageConsumer):
+ self._setup_lti_1p3_ags(consumer)
+ self._setup_lti_1p3_deep_linking(consumer)
+ self._setup_lti_1p3_nrps(consumer)
+
return consumer
def get_lti_consumer(self):
diff --git a/lti_consumer/plugin/urls.py b/lti_consumer/plugin/urls.py
index 39496e8b94283bec37918f2ae51b9c46db4a42b9..69780718c8e77e34a59c6e4737624b8a2b5f5c5a 100644
--- a/lti_consumer/plugin/urls.py
+++ b/lti_consumer/plugin/urls.py
@@ -10,7 +10,8 @@ from rest_framework import routers
from lti_consumer.plugin.views import (LtiAgsLineItemViewset, # LTI Advantage URLs; LTI NRPS URLs
LtiNrpsContextMembershipViewSet, access_token_endpoint,
deep_linking_content_endpoint, deep_linking_response_endpoint,
- launch_gate_endpoint, public_keyset_endpoint)
+ launch_gate_endpoint, public_keyset_endpoint,
+ start_proctoring_assessment_endpoint)
# LTI 1.3 APIs router
router = routers.SimpleRouter(trailing_slash=False)
@@ -59,4 +60,9 @@ urlpatterns = [
r'lti_consumer/v1/lti/(?P<lti_config_id>[-\w]+)/',
include(router.urls)
),
+ path(
+ 'lti_consumer/v1/start_proctoring_assessment',
+ start_proctoring_assessment_endpoint,
+ name='lti_consumer.start_proctoring_assessment_endpoint'
+ ),
]
diff --git a/lti_consumer/plugin/views.py b/lti_consumer/plugin/views.py
index bc896a46d8f362301ca21edcaed4434788a96240..d1f971dc3ce286431927dabc4b0ac48812251fa8 100644
--- a/lti_consumer/plugin/views.py
+++ b/lti_consumer/plugin/views.py
@@ -6,67 +6,48 @@ import urllib
from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist, PermissionDenied, ValidationError
-from django.http import JsonResponse, Http404
from django.db import transaction
+from django.http import Http404, JsonResponse
+from django.shortcuts import render
+from django.utils.crypto import get_random_string
+from django.views.decorators.clickjacking import xframe_options_exempt, xframe_options_sameorigin
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
-from django.views.decorators.clickjacking import xframe_options_exempt, xframe_options_sameorigin
-from django.shortcuts import render
from django_filters.rest_framework import DjangoFilterBackend
+from edx_django_utils.cache import TieredCache, get_cache_key
+from jwkest.jwt import JWT, BadSyntax
from opaque_keys import InvalidKeyError
from opaque_keys.edx.keys import UsageKey
-from rest_framework import viewsets, status
+from rest_framework import status, viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.status import HTTP_400_BAD_REQUEST, HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND
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.models import (
- LtiConfiguration,
- LtiAgsLineItem,
- LtiDlContentItem,
-)
-
-from lti_consumer.lti_1p3.exceptions import (
- Lti1p3Exception,
- LtiDeepLinkingContentTypeNotSupported,
- UnsupportedGrantType,
- MalformedJwtToken,
- MissingRequiredClaim,
- NoSuitableKeys,
- TokenSignatureExpired,
- UnknownClientId,
-)
-from lti_consumer.lti_1p3.extensions.rest_framework.constants import LTI_DL_CONTENT_TYPE_SERIALIZER_MAP
-from lti_consumer.lti_1p3.extensions.rest_framework.serializers import (
- LtiAgsLineItemSerializer,
- LtiAgsScoreSerializer,
- LtiAgsResultSerializer,
- LtiNrpsContextMembershipBasicSerializer,
- LtiNrpsContextMembershipPIISerializer,
-)
-from lti_consumer.lti_1p3.extensions.rest_framework.permissions import (
- LtiAgsPermissions,
- LtiNrpsContextMembershipsPermissions,
-)
+from lti_consumer.lti_1p3.consumer import LtiProctoringConsumer
+from lti_consumer.lti_1p3.exceptions import (BadJwtSignature, InvalidClaimValue, Lti1p3Exception,
+ LtiDeepLinkingContentTypeNotSupported, MalformedJwtToken,
+ MissingRequiredClaim, NoSuitableKeys, TokenSignatureExpired,
+ UnknownClientId, UnsupportedGrantType)
from lti_consumer.lti_1p3.extensions.rest_framework.authentication import Lti1p3ApiAuthentication
-from lti_consumer.lti_1p3.extensions.rest_framework.renderers import (
- LineItemsRenderer,
- LineItemRenderer,
- LineItemScoreRenderer,
- LineItemResultsRenderer,
- MembershipResultRenderer,
-)
-from lti_consumer.lti_1p3.extensions.rest_framework.parsers import (
- LineItemParser,
- LineItemScoreParser,
-)
+from lti_consumer.lti_1p3.extensions.rest_framework.constants import LTI_DL_CONTENT_TYPE_SERIALIZER_MAP
+from lti_consumer.lti_1p3.extensions.rest_framework.parsers import LineItemParser, LineItemScoreParser
+from lti_consumer.lti_1p3.extensions.rest_framework.permissions import (LtiAgsPermissions,
+ LtiNrpsContextMembershipsPermissions)
+from lti_consumer.lti_1p3.extensions.rest_framework.renderers import (LineItemRenderer, LineItemResultsRenderer,
+ LineItemScoreRenderer, LineItemsRenderer,
+ MembershipResultRenderer)
+from lti_consumer.lti_1p3.extensions.rest_framework.serializers import (LtiAgsLineItemSerializer,
+ LtiAgsResultSerializer, LtiAgsScoreSerializer,
+ LtiNrpsContextMembershipBasicSerializer,
+ LtiNrpsContextMembershipPIISerializer)
from lti_consumer.lti_1p3.extensions.rest_framework.utils import IgnoreContentNegotiation
+from lti_consumer.models import LtiAgsLineItem, LtiConfiguration, LtiDlContentItem
from lti_consumer.plugin import compat
-from lti_consumer.utils import _, get_lti_1p3_context_types_claim, get_data_from_cache
+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
log = logging.getLogger(__name__)
@@ -139,13 +120,18 @@ def public_keyset_endpoint(request, usage_id=None, lti_config_id=None):
@require_http_methods(["GET", "POST"])
@xframe_options_exempt
+@csrf_exempt
def launch_gate_endpoint(request, suffix=None): # pylint: disable=unused-argument
"""
- Gate endpoint that triggers LTI launch endpoint XBlock handler
+ Receives an LTI 1.3 authentication request from an LTI tool and returns an LTI 1.3 authentication response.
+ The authentication request and the authentication response are the second and third steps of the OpenID Connect
+ Launch Flow, respectively. The authentication response contains the LTI message and is the LTI launch.
+
+ Returns a response containing an auto-submitting form that directs the browser to make a POST to the Tool.
- This uses the config_id key of the "lti_message_hint" query parameter
- to identify the LtiConfiguration and its consumer to generate the
- LTI 1.3 Launch Form.
+ Query Parameters:
+ * lti_message_hint (REQUIRED): a value used as a cache key to retrieved a cached instance of Lti1p3LaunchData
+ * login_hint (REQUIRED): an identifier for the user that initiated the launch; it is stable and unique to the issuer
"""
# pylint: disable=too-many-statements
request_params = request.GET if request.method == 'GET' else request.POST
@@ -162,7 +148,10 @@ def launch_gate_endpoint(request, suffix=None): # pylint: disable=unused-argume
launch_data = get_data_from_cache(lti_message_hint)
if not launch_data:
- log.warning(f'There was a cache miss during an LTI 1.3 launch when using the cache_key {lti_message_hint}.')
+ log.warning(
+ f'There was a cache miss trying to fetch the launch data during an LTI 1.3 launch when using the cache'
+ f' key {lti_message_hint}. The login hint is {login_hint}.'
+ )
return render(request, 'html/lti_launch_error.html', status=HTTP_400_BAD_REQUEST)
# Validate the Lti1p3LaunchData.
@@ -193,7 +182,7 @@ def launch_gate_endpoint(request, suffix=None): # pylint: disable=unused-argume
lti_consumer = lti_config.get_lti_consumer()
# Set sub and roles claims.
- user_id = launch_data.user_id
+ user_id = launch_data.external_user_id if launch_data.external_user_id else launch_data.user_id
user_role = launch_data.user_role
lti_consumer.set_user_data(
user_id=user_id,
@@ -204,9 +193,10 @@ def launch_gate_endpoint(request, suffix=None): # pylint: disable=unused-argume
lti_consumer.set_resource_link_claim(launch_data.resource_link_id)
# Set launch_presentation claim.
- launch_presentation_target = launch_data.launch_presentation_document_target
- if launch_presentation_target:
- lti_consumer.set_launch_presentation_claim(launch_presentation_target)
+ lti_consumer.set_launch_presentation_claim(
+ document_target=launch_data.launch_presentation_document_target,
+ return_url=launch_data.launch_presentation_return_url
+ )
# Set optional context claim, if supplied.
context_type = launch_data.context_type
@@ -240,7 +230,7 @@ def launch_gate_endpoint(request, suffix=None): # pylint: disable=unused-argume
# course creators to set up content.
deep_linking_content_item_id = launch_data.deep_linking_content_item_id
- if lti_consumer.dl and launch_data.message_type == 'LtiDeepLinkingRequest':
+ if launch_data.message_type == 'LtiDeepLinkingRequest' and lti_consumer.dl:
# Check if the user is staff before LTI doing deep linking launch.
# If not, raise exception and display error page
if user_role not in ['instructor', 'staff']:
@@ -251,7 +241,7 @@ def launch_gate_endpoint(request, suffix=None): # pylint: disable=unused-argume
# Deep Linking ltiResourceLink content presentation
# When content type is `ltiResourceLink`, the tool will be launched with
# different parameters, set by instructors when running the DL configuration flow.
- elif lti_consumer.dl and deep_linking_content_item_id:
+ elif deep_linking_content_item_id and lti_consumer.dl:
# Retrieve Deep Linking parameters using the parameter.
content_item = lti_config.ltidlcontentitem_set.get(pk=deep_linking_content_item_id)
# Only filter DL content item from content item set in the same LTI configuration.
@@ -265,6 +255,30 @@ def launch_gate_endpoint(request, suffix=None): # pylint: disable=unused-argume
custom=dl_params.get('custom')
)
+ if launch_data.message_type == 'LtiStartProctoring':
+ # In the synchronizer token method of CSRF protection, the anti-CSRF token must be stored on the server.
+ session_data_key = get_cache_key(
+ app="lti",
+ key="session_data",
+ user_id=launch_data.user_id,
+ resource_link_id=launch_data.resource_link_id
+ )
+
+ session_data = get_data_from_cache(session_data_key)
+ if not session_data:
+ session_data = get_random_string(32)
+ TieredCache.set_all_tiers(session_data_key, session_data)
+
+ lti_consumer.set_proctoring_data(
+ attempt_number=launch_data.proctoring_launch_data.attempt_number,
+ session_data=session_data,
+ start_assessment_url=launch_data.proctoring_launch_data.start_assessment_url
+ )
+ elif launch_data.message_type == 'LtiEndAssessment':
+ lti_consumer.set_proctoring_data(
+ attempt_number=launch_data.proctoring_launch_data.attempt_number,
+ )
+
# Update context with LTI launch parameters
context.update({
"preflight_response": preflight_response,
@@ -302,6 +316,7 @@ 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):
"""
@@ -471,7 +486,6 @@ def deep_linking_content_endpoint(request, lti_config_id):
if not launch_data:
log.warning(f'There was a cache miss during an LTI 1.3 launch when using the cache_key {launch_data_key}.')
return render(request, 'html/lti_launch_error.html', status=HTTP_400_BAD_REQUEST)
-
try:
# Get LTI Configuration
lti_config = LtiConfiguration.objects.get(id=lti_config_id)
@@ -705,3 +719,113 @@ class LtiNrpsContextMembershipViewSet(viewsets.ReadOnlyModelViewSet):
"error": "above_response_limit",
"explanation": "The number of retrieved users is bigger than the maximum allowed in the configuration.",
}, status=HTTP_403_FORBIDDEN)
+
+
+@csrf_exempt
+@require_http_methods(['POST'])
+def start_proctoring_assessment_endpoint(request):
+ """
+ Receives the Proctoring Tool's message to start the assessment. Emits a signal informing interested parties that
+ the assessment should be started.
+
+ Form Parameters:
+ * JWT (REQUIRED): a signed JWT containing the LTI message
+ """
+ # In order to get the cached data (session_data and launch_data) from the cache, we need data from the JWT
+ # before it has been decoded and validated using the ToolKeyHandler. Grab the data we need and validate the JWT
+ # after.
+ token = request.POST.get('JWT')
+
+ try:
+ jwt = JWT().unpack(token)
+ except BadSyntax:
+ return render(request, 'html/lti_proctoring_start_error.html', status=HTTP_400_BAD_REQUEST)
+
+ jwt_payload = jwt.payload()
+ iss = jwt_payload.get('iss')
+ resource_link_id = jwt_payload.get('https://purl.imsglobal.org/spec/lti/claim/resource_link', {}).get('id')
+
+ try:
+ lti_config = LtiConfiguration.objects.get(lti_1p3_client_id=iss)
+ except LtiConfiguration.DoesNotExist:
+ log.error("Invalid iss claim '%s' for LTI 1.3 Proctoring Services start_proctoring_assessment_endpoint"
+ " callback", iss)
+ return render(request, 'html/lti_proctoring_start_error.html', status=HTTP_404_NOT_FOUND)
+
+ lti_consumer = lti_config.get_lti_consumer()
+
+ if not isinstance(lti_consumer, LtiProctoringConsumer):
+ log.info("Proctoring Services for LTIConfiguration with config_id %s are not enabled", lti_config.config_id)
+ return render(request, 'html/lti_proctoring_start_error.html', status=HTTP_400_BAD_REQUEST)
+
+ # Grab the data we need from the cache: launch_data and session_data.
+ common_cache_key_arguments = {
+ "app": "lti",
+ "user_id": request.user.id,
+ "resource_link_id": resource_link_id,
+ }
+
+ launch_data_key = get_cache_key(**common_cache_key_arguments, key="launch_data")
+ launch_data = get_data_from_cache(launch_data_key)
+ if not launch_data:
+ log.warning(
+ f'There was a cache miss trying to fetch the launch data during an LTI 1.3 proctoring StartAssessment '
+ f'launch when using the cache key {launch_data_key}. The LtiConfiguration config_id is '
+ f'{lti_config.config_id}.'
+ )
+ return render(request, 'html/lti_proctoring_start_error.html', status=HTTP_400_BAD_REQUEST)
+
+ session_data_key = get_cache_key(**common_cache_key_arguments, key="session_data")
+ session_data = get_data_from_cache(session_data_key)
+
+ lti_consumer.set_proctoring_data(
+ attempt_number=launch_data.proctoring_launch_data.attempt_number,
+ session_data=session_data,
+ resource_link_id=launch_data.resource_link_id,
+ )
+
+ try:
+ proctoring_response = lti_consumer.check_and_decode_token(token)
+
+ except (BadJwtSignature, InvalidClaimValue, MalformedJwtToken,
+ MissingRequiredClaim, NoSuitableKeys, TokenSignatureExpired):
+ return render(request, 'html/lti_proctoring_start_error.html', status=HTTP_400_BAD_REQUEST)
+
+ # If the Proctoring Tool specifies the end_assessment_return claim in its LTI launch request,
+ # the Assessment Platform MUST send an End Assessment Message at the end of the user's
+ # proctored exam.
+ end_assessment_return = proctoring_response.get('end_assessment_return')
+ if end_assessment_return:
+ end_assessment_return_key = get_cache_key(**common_cache_key_arguments, key="end_assessment_return")
+ # We convert the boolean to an int because memcached will return an int even if a boolean is stored. This
+ # ensures a consistent return value. This assumes end_assessment_return is a boolean or can otherwise be case to
+ # an integer.
+ try:
+ end_assessment_return_value = int(end_assessment_return)
+ except ValueError:
+ # If the end_assessment_return is not a boolean and cannot be cast to an integer, then assume that the value
+ # is False. We do not want to return a 404 at the end of a proctored session on account of an invalid value
+ # for this optional claim.
+ log.error(
+ "An error occurred during the handling of an LtiStartAssessment LTI lauch message for LTIConfiguration "
+ f"with config_id {lti_config.config_id} and resource_link_id {resource_link_id}. The "
+ "end_assessment_return Tool JWT claim is not a boolean value. An LtiEndAssessment LTI launch message "
+ "will not be sent as part of the end assessment workflow."
+ )
+ else:
+ # Set a long enough timeout to ensure learners can complete their assessments without a cache timeout.
+ timeout = 60 * 60 * 12
+ TieredCache.set_all_tiers(
+ end_assessment_return_key,
+ end_assessment_return_value,
+ django_cache_timeout=timeout
+ )
+
+ LTI_1P3_PROCTORING_ASSESSMENT_STARTED.send(
+ sender=None,
+ attempt_number=proctoring_response["attempt_number"],
+ resource_link=proctoring_response["resource_link"],
+ user_id=request.user.id,
+ )
+
+ return JsonResponse(data={})
diff --git a/lti_consumer/signals/__init__.py b/lti_consumer/signals/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
diff --git a/lti_consumer/signals.py b/lti_consumer/signals/signals.py
similarity index 96%
rename from lti_consumer/signals.py
rename to lti_consumer/signals/signals.py
index e9d928e9faea722b96f05febe3f48d4cb1cd6082..ccf1194fed9e31aa2cb6487bdb132f3e309c911e 100644
--- a/lti_consumer/signals.py
+++ b/lti_consumer/signals/signals.py
@@ -4,7 +4,7 @@ LTI Consumer related Signal handlers
import logging
from django.db.models.signals import post_save
-from django.dispatch import receiver
+from django.dispatch import receiver, Signal
from lti_consumer.models import LtiAgsScore
from lti_consumer.plugin import compat
@@ -62,3 +62,6 @@ def publish_grade_on_score_update(sender, instance, **kwargs): # pylint: disabl
exc,
)
raise exc
+
+
+LTI_1P3_PROCTORING_ASSESSMENT_STARTED = Signal()
diff --git a/lti_consumer/templates/html/lti_proctoring_start_error.html b/lti_consumer/templates/html/lti_proctoring_start_error.html
new file mode 100644
index 0000000000000000000000000000000000000000..21d9eeaad9d9b785c61cea33694664cbd76e1664
--- /dev/null
+++ b/lti_consumer/templates/html/lti_proctoring_start_error.html
@@ -0,0 +1,16 @@
+{% load i18n %}
+<!DOCTYPE HTML>
+<html>
+ <head>
+ <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
+ <title>LTI</title>
+ </head>
+ <body>
+ <p>
+ <b>{% trans "There was an error while starting your LTI proctored assessment." %}</b>
+ </p>
+ <p>
+ {% trans "If you're seeing this on a live course, please contact the course staff." %}
+ </p>
+ </body>
+</html>
diff --git a/lti_consumer/tests/unit/plugin/test_proctoring.py b/lti_consumer/tests/unit/plugin/test_proctoring.py
new file mode 100644
index 0000000000000000000000000000000000000000..25a62456ba3e0f6463d1650c8db146396758110d
--- /dev/null
+++ b/lti_consumer/tests/unit/plugin/test_proctoring.py
@@ -0,0 +1,282 @@
+""""
+Tests for LTI 1.3 proctoring endpoint views.
+"""
+import uuid
+from unittest.mock import call, patch
+
+import ddt
+from Cryptodome.PublicKey import RSA
+from django.contrib.auth import get_user_model
+from django.test.testcases import TestCase
+from edx_django_utils.cache import TieredCache, get_cache_key
+from jwkest.jwk import RSAKey
+from jwkest.jwt import BadSyntax
+
+from lti_consumer.data import Lti1p3LaunchData, Lti1p3ProctoringLaunchData
+from lti_consumer.lti_1p3.exceptions import (BadJwtSignature, InvalidClaimValue, MalformedJwtToken,
+ MissingRequiredClaim, NoSuitableKeys)
+from lti_consumer.lti_1p3.key_handlers import PlatformKeyHandler
+from lti_consumer.models import LtiConfiguration
+from lti_consumer.utils import get_data_from_cache
+
+
+@ddt.ddt
+class TestLti1p3ProctoringStartProctoringAssessmentEndpoint(TestCase):
+ """Tests for the start_proctoring_assessment_endpoint endpoint."""
+
+ def setUp(self):
+ super().setUp()
+
+ self.url = "/lti_consumer/v1/start_proctoring_assessment"
+
+ # Set up user.
+ self._setup_user()
+
+ # Set up an LtiConfiguration instance for the integration.
+ self.lti_config = LtiConfiguration.objects.create(
+ version=LtiConfiguration.LTI_1P3,
+ lti_1p3_proctoring_enabled=True,
+ config_store=LtiConfiguration.CONFIG_ON_DB,
+ )
+
+ # Set up cached data necessary for this endpoint: launch_data and session_data.
+ self._setup_cached_data()
+
+ # Set up a public key - private key pair that allows encoding and decoding a Tool JWT.
+ self.rsa_key_id = str(uuid.uuid4())
+ self.private_key = RSA.generate(2048)
+ self.key = RSAKey(
+ key=self.private_key,
+ kid=self.rsa_key_id
+ )
+ self.public_key = self.private_key.publickey().export_key().decode()
+
+ self.lti_config.lti_1p3_tool_public_key = self.public_key
+ self.lti_config.save()
+
+ def _setup_user(self):
+ """Sets up the requesting user instance."""
+ self.user = get_user_model().objects.create_user(
+ username="user",
+ password="password"
+ )
+ self.client.login(username="user", password="password")
+
+ def _setup_cached_data(self):
+ """Sets up data in the cache necessary for the view: launch_data and session_data."""
+ self.common_cache_key_arguments = {
+ "app": "lti",
+ "user_id": self.user.id,
+ "resource_link_id": "resource_link_id",
+ }
+
+ # Cache session_data.
+ self.session_data_key = get_cache_key(
+ **self.common_cache_key_arguments,
+ key="session_data"
+ )
+ TieredCache.set_all_tiers(self.session_data_key, "session_data")
+
+ # Cache launch_data.
+ proctoring_launch_data = Lti1p3ProctoringLaunchData(attempt_number=2)
+ launch_data = Lti1p3LaunchData(
+ user_id="1",
+ user_role=None,
+ config_id=self.lti_config.config_id,
+ resource_link_id="resource_link_id",
+ proctoring_launch_data=proctoring_launch_data,
+ )
+
+ self.launch_data_key = get_cache_key(
+ **self.common_cache_key_arguments,
+ key="launch_data"
+ )
+ TieredCache.set_all_tiers(self.launch_data_key, launch_data)
+
+ def create_tool_jwt_token(self, **kwargs):
+ """
+ Creates and returns a signed JWT token to act as a Tool JWT.
+
+ Arguments:
+ * kwargs: Keyword arguments representing key, value pairs to include in the JWT token. This allows callers
+ to override default claims in the JWT token.
+ """
+ lti_consumer = self.lti_config.get_lti_consumer()
+
+ token = {
+ "iss": lti_consumer.client_id,
+ "https://purl.imsglobal.org/spec/lti/claim/message_type": "LtiStartAssessment",
+ "https://purl.imsglobal.org/spec/lti/claim/version": "1.3.0",
+ "https://purl.imsglobal.org/spec/lti-ap/claim/session_data": "session_data",
+ "https://purl.imsglobal.org/spec/lti/claim/resource_link": {"id": "resource_link_id"},
+ "https://purl.imsglobal.org/spec/lti-ap/claim/attempt_number": 2,
+ }
+
+ token.update(**kwargs)
+
+ # Encode and sign the Tool JWT using the private key. The PlatformKeyHandler class is the only key handler that
+ # currently has code to encode and sign a JWT, so we use that class. The ToolKeyHandler will be used in the view
+ # to decode this JWT using the corresponding public key.
+ platform_key_handler = PlatformKeyHandler(self.private_key.export_key(), self.rsa_key_id)
+ signed_token = platform_key_handler.encode_and_sign(token)
+
+ return signed_token
+
+ def test_valid_token(self):
+ """Tests the happy path of the start_proctoring_assessment_endpoint."""
+ response = self.client.post(
+ self.url,
+ {
+ "JWT": self.create_tool_jwt_token()
+ },
+ )
+ self.assertEqual(response.status_code, 200)
+
+ def test_unparsable_token(self):
+ """Tests that a call to the start_assessment_endpoint with an unparsable token results in a 400 response."""
+ with patch("lti_consumer.plugin.views.JWT.unpack") as mock_jwt_unpack_method:
+ mock_jwt_unpack_method.side_effect = BadSyntax(value="", msg="")
+
+ response = self.client.post(
+ self.url,
+ {
+ "JWT": self.create_tool_jwt_token()
+ },
+ )
+ self.assertEqual(response.status_code, 400)
+
+ def test_lti_configuration_does_not_exist(self):
+ """
+ Tests that a call to the start_assessment_endpoint with an "iss" Tool JWT token claim that does not correspond
+ to an LtiConfiguration instance results in a 404 response.
+ """
+ tool_jwt_token_overrides = {"iss": "iss"}
+ tool_jwt_token = self.create_tool_jwt_token(**tool_jwt_token_overrides)
+
+ response = self.client.post(
+ self.url,
+ {
+ "JWT": tool_jwt_token
+ },
+ )
+ self.assertEqual(response.status_code, 404)
+
+ def test_not_proctoring_consumer(self):
+ """
+ Tests that a call to the start_assessment_endpoint with an "iss" Tool JWT token claim that corresponds to
+ an LtiConfiguration instance that does not have proctoring enabled results in a 400 response.
+ """
+ # Disable LTI Assessment and Grades Services so we don't set it up unnecessarily. Otherwise, an exception is
+ # raised because there is no location field set on the LtiConfiguration instance.
+ with patch("lti_consumer.models.LtiConfiguration.get_lti_advantage_ags_mode") as get_lti_ags_mode_mock:
+ get_lti_ags_mode_mock.return_value = self.lti_config.LTI_ADVANTAGE_AGS_DISABLED
+ self.lti_config.lti_1p3_proctoring_enabled = False
+ self.lti_config.save()
+
+ response = self.client.post(
+ self.url,
+ {
+ "JWT": self.create_tool_jwt_token()
+ },
+ )
+ self.assertEqual(response.status_code, 400)
+
+ def test_cache_miss_launch_data(self):
+ """Tests that a call to the start_assessment_endpoint with no cached launch_data results in a 400 response."""
+ TieredCache.set_all_tiers(self.launch_data_key, None)
+
+ response = self.client.post(
+ self.url,
+ {
+ "JWT": self.create_tool_jwt_token()
+ },
+ )
+ self.assertEqual(response.status_code, 400)
+
+ @ddt.data(BadJwtSignature, InvalidClaimValue, MalformedJwtToken, MissingRequiredClaim, NoSuitableKeys)
+ def test_check_and_decode_token_exception_handling(self, exception):
+ """Tests that a call to the start_assessment_endpoint with an invalid token results in a 400 response."""
+ with patch("lti_consumer.lti_1p3.consumer.LtiProctoringConsumer.check_and_decode_token") as mock_method:
+ mock_method.side_effect = exception()
+
+ response = self.client.post(
+ self.url,
+ {
+ "JWT": self.create_tool_jwt_token()
+ },
+ )
+ self.assertEqual(response.status_code, 400)
+
+ def test_cached_end_assessment_return_valid(self):
+ """
+ Tests that a call to the start_assessment_endpoint with a valid end_assessment_return Tool JWT token claim
+ results in a 200 response and correct data in the cache.
+ """
+ end_assessment_return = True
+
+ tool_jwt_token_overrides = {
+ "https://purl.imsglobal.org/spec/lti-ap/claim/end_assessment_return": end_assessment_return,
+ }
+ tool_jwt_token = self.create_tool_jwt_token(**tool_jwt_token_overrides)
+
+ response = self.client.post(
+ self.url,
+ {
+ "JWT": tool_jwt_token
+ },
+ )
+
+ self.assertEqual(response.status_code, 200)
+
+ end_assessment_return_cache_key = get_cache_key(**self.common_cache_key_arguments, key="end_assessment_return")
+ end_assessment_return = get_data_from_cache(end_assessment_return_cache_key)
+ self.assertEqual(end_assessment_return, int(end_assessment_return))
+
+ def test_cached_end_assessment_return_invalid(self):
+ """
+ Tests that a call to the start_assessment_endpoint with an invalid end_assessment_return Tool JWT token claim
+ results in a 200 response and correct data in the cache.
+ """
+ end_assessment_return = "end_assessment_return"
+
+ tool_jwt_token_overrides = {
+ "https://purl.imsglobal.org/spec/lti-ap/claim/end_assessment_return": end_assessment_return,
+ }
+ tool_jwt_token = self.create_tool_jwt_token(**tool_jwt_token_overrides)
+
+ response = self.client.post(
+ self.url,
+ {
+ "JWT": tool_jwt_token
+ },
+ )
+
+ self.assertEqual(response.status_code, 200)
+
+ end_assessment_return_cache_key = get_cache_key(**self.common_cache_key_arguments, key="end_assessment_return")
+ end_assessment_return = get_data_from_cache(end_assessment_return_cache_key)
+ self.assertEqual(end_assessment_return, None)
+
+ @patch("lti_consumer.plugin.views.LTI_1P3_PROCTORING_ASSESSMENT_STARTED.send")
+ def test_lti_1p3_proctoring_assessment_started_signal(self, mock_assessment_started_signal):
+ """
+ Tests that a successful call to the start_assessment_endpoint emits the LTI_1P3_PROCTORING_ASSESSMENT_STARTED
+ Django signal.
+ """
+ self.client.post(
+ self.url,
+ {
+ "JWT": self.create_tool_jwt_token()
+ },
+ )
+
+ self.assertTrue(mock_assessment_started_signal.called)
+ self.assertEqual(mock_assessment_started_signal.call_count, 1)
+
+ expected_call_args = call(
+ sender=None,
+ attempt_number=2,
+ resource_link={'id': 'resource_link_id'},
+ user_id=self.user.id,
+ )
+ self.assertEqual(mock_assessment_started_signal.call_args, expected_call_args)
diff --git a/lti_consumer/tests/unit/plugin/test_views.py b/lti_consumer/tests/unit/plugin/test_views.py
index e1bce12dfd4c6de283d42f487a68301ec7200e0c..7c6740327ddd0c4433d331f949cd1356a40d607e 100644
--- a/lti_consumer/tests/unit/plugin/test_views.py
+++ b/lti_consumer/tests/unit/plugin/test_views.py
@@ -8,11 +8,12 @@ import ddt
from django.test.testcases import TestCase
from django.urls import reverse
+from edx_django_utils.cache import TieredCache, get_cache_key
from Cryptodome.PublicKey import RSA
from jwkest.jwk import RSAKey
from opaque_keys.edx.keys import UsageKey
-from lti_consumer.data import Lti1p3LaunchData
+from lti_consumer.data import Lti1p3LaunchData, Lti1p3ProctoringLaunchData
from lti_consumer.models import LtiConfiguration, LtiDlContentItem
from lti_consumer.lti_1p3.exceptions import (
MissingRequiredClaim,
@@ -456,6 +457,72 @@ class TestLti1p3LaunchGateEndpoint(TestCase):
# Check response
self.assertEqual(response.status_code, 200)
+ def test_launch_callback_endpoint_start_proctoring(self):
+ """
+ Ensures that the launch_callback_endpoint works correctly for LtiStartProctoring LTI launch messages.
+ """
+ self.config.lti_1p3_proctoring_enabled = True
+ self.config.save()
+
+ self.launch_data.message_type = "LtiStartProctoring"
+
+ proctoring_launch_data = Lti1p3ProctoringLaunchData(
+ attempt_number=1,
+ start_assessment_url="start_assessment_url",
+ )
+
+ self.launch_data.proctoring_launch_data = proctoring_launch_data
+
+ session_data_key = get_cache_key(
+ app="lti",
+ key="session_data",
+ user_id=self.launch_data.user_id,
+ resource_link_id=self.launch_data.resource_link_id
+ )
+
+ TieredCache.set_all_tiers(session_data_key, "session_data")
+
+ 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)
+
+ # Check response
+ self.assertEqual(response.status_code, 200)
+
+ def test_launch_callback_endpoint_end_assessment(self):
+ """
+ Ensures that the launch_callback_endpoint works correctly for LtiEndAssessment LTI launch messages.
+ """
+ self.config.lti_1p3_proctoring_enabled = True
+ self.config.save()
+
+ self.launch_data.message_type = "LtiEndAssessment"
+
+ proctoring_launch_data = Lti1p3ProctoringLaunchData(
+ attempt_number=1,
+ )
+
+ self.launch_data.proctoring_launch_data = proctoring_launch_data
+
+ 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)
+
+ # Check response
+ self.assertEqual(response.status_code, 200)
+
class TestLti1p3AccessTokenEndpoint(TestCase):
"""
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 a3dd32845daa67c2d345f1889b96758cbefa42a7..a80a142fb9bb365454396a44973a5f5681312824 100644
--- a/lti_consumer/tests/unit/plugin/test_views_lti_ags.py
+++ b/lti_consumer/tests/unit/plugin/test_views_lti_ags.py
@@ -64,7 +64,7 @@ class LtiAgsLineItemViewSetTestCase(APITransactionTestCase):
self._load_block_patch.return_value = self.xblock
self._mock_user = Mock()
- compat_mock = patch("lti_consumer.signals.compat")
+ compat_mock = patch("lti_consumer.signals.signals.compat")
self.addCleanup(compat_mock.stop)
self._compat_mock = compat_mock.start()
self._compat_mock.get_user_from_external_user_id.return_value = self._mock_user
diff --git a/lti_consumer/tests/unit/plugin/test_views_lti_deep_linking.py b/lti_consumer/tests/unit/plugin/test_views_lti_deep_linking.py
index 6f1c5eeb956f8efad7f979c86293ce75e8ee564d..5d29d12dd0529d5ffdc4865657d71dcf2e07d109 100644
--- a/lti_consumer/tests/unit/plugin/test_views_lti_deep_linking.py
+++ b/lti_consumer/tests/unit/plugin/test_views_lti_deep_linking.py
@@ -96,7 +96,7 @@ class LtiDeepLinkingResponseEndpointTestCase(LtiDeepLinkingTestCase):
super().setUp()
# Patch method that calls platform core to ask for user permissions
- compat_mock = patch("lti_consumer.signals.compat")
+ compat_mock = patch("lti_consumer.signals.signals.compat")
self.addCleanup(compat_mock.stop)
self._compat_mock = compat_mock.start()
self._compat_mock.user_has_studio_write_access.return_value = True
diff --git a/lti_consumer/tests/unit/test_api.py b/lti_consumer/tests/unit/test_api.py
index 446e07743ec9b09728a9dacfdc27f033662b71d8..19ff9f2f7292e48b9438cde263c77486e2195675 100644
--- a/lti_consumer/tests/unit/test_api.py
+++ b/lti_consumer/tests/unit/test_api.py
@@ -13,13 +13,14 @@ from lti_consumer.api import (
_get_config_by_config_id,
_get_or_create_local_lti_config,
config_id_for_block,
+ get_end_assessment_return,
get_lti_1p3_content_url,
get_deep_linking_data,
get_lti_1p3_launch_info,
get_lti_1p3_launch_start_url,
validate_lti_1p3_launch_data,
)
-from lti_consumer.data import Lti1p3LaunchData
+from lti_consumer.data import Lti1p3LaunchData, Lti1p3ProctoringLaunchData
from lti_consumer.lti_xblock import LtiConsumerXBlock
from lti_consumer.models import LtiConfiguration, LtiDlContentItem
from lti_consumer.tests.test_utils import make_xblock
@@ -93,6 +94,8 @@ class TestConfigIdForBlock(TestCase):
creation forks on store type.
"""
def setUp(self):
+ super().setUp()
+
xblock_attributes = {
'lti_version': LtiConfiguration.LTI_1P1,
}
@@ -236,6 +239,7 @@ class TestGetOrCreateLocalLtiConfiguration(TestCase):
self.assertEqual(lti_config.external_id, None)
+@ddt.ddt
class TestValidateLti1p3LaunchData(TestCase):
"""
Unit tests for validate_lti_1p3_launch_data API method.
@@ -306,11 +310,11 @@ class TestValidateLti1p3LaunchData(TestCase):
self.assertEqual(is_valid, False)
self._assert_required_context_id_message(validation_messages)
- def test_invalid_user_role(self):
+ @ddt.data("cat", "")
+ def test_invalid_user_role(self, user_role):
"""
Ensure that instances of Lti1p3LaunchData are instantiated with a user_role that is in the LTI_1P3_ROLE_MAP.
"""
- user_role = "cat"
launch_data = Lti1p3LaunchData(
user_id="1",
user_role=user_role,
@@ -326,6 +330,22 @@ class TestValidateLti1p3LaunchData(TestCase):
[f"The user_role attribute {user_role} is not a valid user_role."]
)
+ def test_none_user_role(self):
+ """
+ Ensure that instances of Lti1p3LaunchData can be instantiated with a value of None for user_role.
+ """
+ launch_data = Lti1p3LaunchData(
+ user_id="1",
+ user_role=None,
+ config_id=_test_config_id,
+ resource_link_id="resource_link_id",
+ )
+
+ is_valid, validation_messages = validate_lti_1p3_launch_data(launch_data)
+
+ self.assertEqual(is_valid, True)
+ self.assertEqual(validation_messages, [])
+
def test_invalid_context_type(self):
"""
Ensure that instances of Lti1p3LaunchData are instantiated with a context_type that is one of group,
@@ -350,6 +370,51 @@ class TestValidateLti1p3LaunchData(TestCase):
[f"The context_type attribute {context_type} in the launch data is not a valid context_type."]
)
+ @ddt.data("LtiStartProctoring", "LtiEndAssessment")
+ def test_required_proctoring_launch_data_for_proctoring_message_type(self, message_type):
+ launch_data = Lti1p3LaunchData(
+ user_id="1",
+ user_role="student",
+ config_id=_test_config_id,
+ resource_link_id="resource_link_id",
+ message_type=message_type
+ )
+
+ is_valid, validation_messages = validate_lti_1p3_launch_data(launch_data)
+
+ self.assertEqual(is_valid, False)
+ self.assertEqual(
+ validation_messages,
+ [
+ "The proctoring_launch_data attribute is required if the message_type attribute is "
+ "\"LtiStartProctoring\" or \"LtiEndAssessment\"."
+ ]
+ )
+
+ @ddt.data(None, "")
+ def test_required_start_assessment_url_for_start_proctoring_message_type(self, start_assessment_url):
+ proctoring_launch_data = Lti1p3ProctoringLaunchData(attempt_number=1, start_assessment_url=start_assessment_url)
+
+ launch_data = Lti1p3LaunchData(
+ user_id="1",
+ user_role="student",
+ config_id=_test_config_id,
+ resource_link_id="resource_link_id",
+ message_type="LtiStartProctoring",
+ proctoring_launch_data=proctoring_launch_data,
+ )
+
+ is_valid, validation_messages = validate_lti_1p3_launch_data(launch_data)
+
+ self.assertEqual(is_valid, False)
+ self.assertEqual(
+ validation_messages,
+ [
+ "The proctoring_start_assessment_url attribute is required if the message_type attribute is"
+ " \"LtiStartProctoring\"."
+ ]
+ )
+
class TestGetLti1p3LaunchInfo(TestCase):
"""
@@ -648,3 +713,26 @@ class TestGetLtiDlContentItemData(TestCase):
with self.assertRaises(Exception):
get_deep_linking_data(content_item.id, self.lti_config.config_id)
+
+
+class TestGetEndAssessmentReturn(TestCase):
+ """
+ Unit tests for get_end_assessment_return API method.
+ """
+ def setUp(self):
+ # Patch internal method to avoid calls to modulestore
+ super().setUp()
+ patcher = patch(
+ 'lti_consumer.models.LtiConfiguration.get_lti_consumer',
+ )
+ self.addCleanup(patcher.stop)
+
+ @patch('lti_consumer.api.get_data_from_cache')
+ def test_get_end_assessment_return(self, mock_get_data_from_cache):
+ """Ensures get_end_assessment_return returns whatever is in the cache."""
+
+ get_data_from_cache_return_value = "end_assessment_return"
+
+ mock_get_data_from_cache.return_value = get_data_from_cache_return_value
+
+ self.assertEqual(get_end_assessment_return("user_id", "resource_link_id"), get_data_from_cache_return_value)
diff --git a/lti_consumer/tests/unit/test_lti_xblock.py b/lti_consumer/tests/unit/test_lti_xblock.py
index adbf3d2e533caa539760682fc942e094e8599e42..4824c8b29cb8541a3ad1dc05dd716d385c90f52e 100644
--- a/lti_consumer/tests/unit/test_lti_xblock.py
+++ b/lti_consumer/tests/unit/test_lti_xblock.py
@@ -257,11 +257,11 @@ class TestProperties(TestLtiConsumerXBlock):
"""
fake_user = Mock()
fake_user.opt_attrs = {
- 'edx-platform.anonymous_user_id': FAKE_USER_ID
+ 'edx-platform.user_id': FAKE_USER_ID
}
self.xblock.runtime.service(self, 'user').get_current_user = Mock(return_value=fake_user)
- self.assertEqual(self.xblock.user_id, FAKE_USER_ID)
+ self.assertEqual(self.xblock.lms_user_id, FAKE_USER_ID)
def test_user_id_none(self):
"""
@@ -275,7 +275,7 @@ class TestProperties(TestLtiConsumerXBlock):
self.xblock.runtime.service(self, 'user').get_current_user = Mock(return_value=fake_user)
with self.assertRaises(LtiError):
- __ = self.xblock.user_id
+ __ = self.xblock.lms_user_id
def test_external_user_id(self):
"""
@@ -304,7 +304,7 @@ class TestProperties(TestLtiConsumerXBlock):
@patch('lti_consumer.lti_xblock.LtiConsumerXBlock.context_id')
@patch('lti_consumer.lti_xblock.LtiConsumerXBlock.resource_link_id')
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.user_id', PropertyMock(return_value=FAKE_USER_ID))
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.anonymous_user_id', PropertyMock(return_value=FAKE_USER_ID))
def test_lis_result_sourcedid(self, mock_resource_link_id, mock_context_id):
"""
Test `lis_result_sourcedid` returns appropriate string
@@ -784,7 +784,7 @@ class TestLtiLaunchHandler(TestLtiConsumerXBlock):
self.xblock.runtime.service(self, 'user').get_current_user = Mock(return_value=fake_user)
@patch('lti_consumer.lti_xblock.LtiConsumerXBlock.course')
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.user_id', PropertyMock(return_value=FAKE_USER_ID))
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.anonymous_user_id', PropertyMock(return_value=FAKE_USER_ID))
def test_generate_launch_request_called(self, mock_course):
"""
Test LtiConsumer.generate_launch_request is called and a 200 HTML response is returned
@@ -832,7 +832,7 @@ class TestLtiLaunchHandler(TestLtiConsumerXBlock):
self.assertIn("There was an error while launching the LTI tool.", response_body)
@patch('lti_consumer.lti_xblock.LtiConsumerXBlock.course')
- @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.user_id', PropertyMock(return_value=FAKE_USER_ID))
+ @patch('lti_consumer.lti_xblock.LtiConsumerXBlock.anonymous_user_id', PropertyMock(return_value=FAKE_USER_ID))
def test_publish_tracking_event(self, mock_course):
"""
Test a tracking event is emitted when generating a launch request
@@ -1472,6 +1472,7 @@ class TestLtiConsumer1p3XBlock(TestCase):
# Mock out the user role and external_user_id properties.
fake_user = Mock()
fake_user.opt_attrs = {
+ 'edx-platform.user_id': 1,
'edx-platform.user_role': 'instructor',
'edx-platform.is_authenticated': True,
}
@@ -1485,10 +1486,11 @@ class TestLtiConsumer1p3XBlock(TestCase):
course_key = str(self.xblock.location.course_key) # pylint: disable=no-member
expected_launch_data = Lti1p3LaunchData(
- user_id="external_user_id",
+ user_id=1,
user_role="instructor",
config_id=config_id_for_block(self.xblock),
resource_link_id=str(self.xblock.location), # pylint: disable=no-member
+ external_user_id="external_user_id",
launch_presentation_document_target="iframe",
message_type="LtiResourceLinkRequest",
context_id=course_key,
diff --git a/lti_consumer/tests/unit/test_models.py b/lti_consumer/tests/unit/test_models.py
index e6d1298e3aad65eb5588e8423e7843022cd2595b..f9b189c9eee6e14b298e0c23c5d2627604df7b77 100644
--- a/lti_consumer/tests/unit/test_models.py
+++ b/lti_consumer/tests/unit/test_models.py
@@ -15,13 +15,8 @@ from jwkest.jwk import RSAKey
from opaque_keys.edx.locator import CourseLocator
from lti_consumer.lti_xblock import LtiConsumerXBlock
-from lti_consumer.models import (
- CourseAllowPIISharingInLTIFlag,
- LtiAgsLineItem,
- LtiAgsScore,
- LtiConfiguration,
- LtiDlContentItem,
-)
+from lti_consumer.models import (CourseAllowPIISharingInLTIFlag, LtiAgsLineItem, LtiAgsScore, LtiConfiguration,
+ LtiDlContentItem)
from lti_consumer.tests.test_utils import make_xblock
@@ -375,6 +370,13 @@ class TestLtiConfigurationModel(TestCase):
self.assertRaises(ValidationError):
self.lti_1p3_config_db.clean()
+ self.lti_1p3_config.lti_1p3_proctoring_enabled = True
+
+ for config_store in [self.lti_1p3_config.CONFIG_ON_XBLOCK, self.lti_1p3_config.CONFIG_EXTERNAL]:
+ self.lti_1p3_config.config_store = config_store
+ with self.assertRaises(ValidationError):
+ self.lti_1p3_config.clean()
+
class TestLtiAgsLineItemModel(TestCase):
"""
@@ -411,7 +413,7 @@ class TestLtiAgsScoreModel(TestCase):
super().setUp()
# patch things related to LtiAgsScore post_save signal receiver
- compat_mock = patch("lti_consumer.signals.compat")
+ compat_mock = patch("lti_consumer.signals.signals.compat")
self.addCleanup(compat_mock.stop)
self._compat_mock = compat_mock.start()
self._compat_mock.load_block_as_user.return_value = make_xblock(
diff --git a/lti_consumer/utils.py b/lti_consumer/utils.py
index 2514c627a7b2d2e1e52a9e456a58d24decc2648d..9413dcd8bd69d55b2e09fa6de6420b44b8bb3e65 100644
--- a/lti_consumer/utils.py
+++ b/lti_consumer/utils.py
@@ -10,6 +10,7 @@ from edx_django_utils.cache import get_cache_key, TieredCache
from lti_consumer.plugin.compat import get_external_config_waffle_flag, get_database_config_waffle_flag
from lti_consumer.lti_1p3.constants import LTI_1P3_CONTEXT_TYPE
+from lti_consumer.lti_1p3.exceptions import InvalidClaimValue, MissingRequiredClaim
log = logging.getLogger(__name__)
@@ -271,3 +272,20 @@ def get_data_from_cache(cache_key):
return cached_data.value
return None
+
+
+def check_token_claim(token, claim_key, expected_value=None, invalid_claim_error_msg=None):
+ """
+ Checks that the claim with key claim_key appears in the token. Raises a MissingRequiredClaim exception if it does
+ not. If the optional arguments expected_value and invalid_claim_error_msg are provided, then checks that the claim
+ in the token with the key claim_key matches the expected_value. Raises an InvalidClaimValue exception with the
+ invalid_claim_error_msg as the message if not. If the invalid_claim_error_msg argument is provided, then a generic
+ message is used.
+ """
+ claim_value = token.get(claim_key)
+
+ if claim_value is None:
+ raise MissingRequiredClaim(f"Token is missing required {claim_key} claim.")
+ if expected_value and claim_value != expected_value:
+ msg = invalid_claim_error_msg if invalid_claim_error_msg else f"The claim {claim_key} value is invalid."
+ raise InvalidClaimValue(msg)