Skip to content

E
edx-platform
Commit f03a80c2 by David Ormsbee
Options

Merge pull request #8205 from mcgachey/mcgachey-lti-grade-passback 

[LTI Provider] Use the LTI Outcome Service to pass back scores
parents 15291aa4 0fe41374
Showing with 595 additions and 30 deletions
lms/djangoapps/lti_provider/__init__.py
"""
The LTI Provider app gives a way to launch edX content via a campus LMS
platform. LTI is a standard protocol for connecting educational tools, defined
by IMS:
http://www.imsglobal.org/toolsinteroperability2.cfm
"""
# Import the tasks module to ensure that signal handlers are registered.
import lms.djangoapps.lti_provider.tasks
lms/djangoapps/lti_provider/migrations/0002_create_lti_outcome_management.py 0 → 100644
# -*- coding: utf-8 -*-
# pylint: disable=invalid-name, missing-docstring, unused-argument, unused-import, line-too-long
from south.utils import datetime_utils as datetime
from south.db import db
from south.v2 import SchemaMigration
from django.db import models
class Migration(SchemaMigration):
def forwards(self, orm):
# Adding model 'OutcomeService'
db.create_table('lti_provider_outcomeservice', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('lis_outcome_service_url', self.gf('django.db.models.fields.CharField')(unique=True, max_length=255)),
('lti_consumer', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['lti_provider.LtiConsumer'])),
))
db.send_create_signal('lti_provider', ['OutcomeService'])
# Adding model 'GradedAssignment'
db.create_table('lti_provider_gradedassignment', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('user', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['auth.User'])),
('course_key', self.gf('xmodule_django.models.CourseKeyField')(max_length=255, db_index=True)),
('usage_key', self.gf('xmodule_django.models.UsageKeyField')(max_length=255, db_index=True)),
('outcome_service', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['lti_provider.OutcomeService'])),
('lis_result_sourcedid', self.gf('django.db.models.fields.CharField')(max_length=255, db_index=True)),
))
db.send_create_signal('lti_provider', ['GradedAssignment'])
# Adding unique constraint on 'GradedAssignment', fields ['outcome_service', 'lis_result_sourcedid']
db.create_unique('lti_provider_gradedassignment', ['outcome_service_id', 'lis_result_sourcedid'])
# Adding field 'LtiConsumer.instance_guid'
db.add_column('lti_provider_lticonsumer', 'instance_guid',
self.gf('django.db.models.fields.CharField')(max_length=255, null=True),
keep_default=False)
# Adding unique constraint on 'LtiConsumer', fields ['consumer_name']
db.create_unique('lti_provider_lticonsumer', ['consumer_name'])
def backwards(self, orm):
# Removing unique constraint on 'LtiConsumer', fields ['consumer_name']
db.delete_unique('lti_provider_lticonsumer', ['consumer_name'])
# Removing unique constraint on 'GradedAssignment', fields ['outcome_service', 'lis_result_sourcedid']
db.delete_unique('lti_provider_gradedassignment', ['outcome_service_id', 'lis_result_sourcedid'])
# Deleting model 'OutcomeService'
db.delete_table('lti_provider_outcomeservice')
# Deleting model 'GradedAssignment'
db.delete_table('lti_provider_gradedassignment')
# Deleting field 'LtiConsumer.instance_guid'
db.delete_column('lti_provider_lticonsumer', 'instance_guid')
models = {
'auth.group': {
'Meta': {'object_name': 'Group'},
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
},
'auth.permission': {
'Meta': {'ordering': "('content_type__app_label', 'content_type__model', 'codename')", 'unique_together': "(('content_type', 'codename'),)", 'object_name': 'Permission'},
'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
},
'auth.user': {
'Meta': {'object_name': 'User'},
'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'symmetrical': 'False', 'blank': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'}),
'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
},
'contenttypes.contenttype': {
'Meta': {'ordering': "('name',)", 'unique_together': "(('app_label', 'model'),)", 'object_name': 'ContentType', 'db_table': "'django_content_type'"},
'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
},
'lti_provider.gradedassignment': {
'Meta': {'unique_together': "(('outcome_service', 'lis_result_sourcedid'),)", 'object_name': 'GradedAssignment'},
'course_key': ('xmodule_django.models.CourseKeyField', [], {'max_length': '255', 'db_index': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'lis_result_sourcedid': ('django.db.models.fields.CharField', [], {'max_length': '255', 'db_index': 'True'}),
'outcome_service': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['lti_provider.OutcomeService']"}),
'usage_key': ('xmodule_django.models.UsageKeyField', [], {'max_length': '255', 'db_index': 'True'}),
'user': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']"})
},
'lti_provider.lticonsumer': {
'Meta': {'object_name': 'LtiConsumer'},
'consumer_key': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '32', 'db_index': 'True'}),
'consumer_name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '255'}),
'consumer_secret': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '32'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'instance_guid': ('django.db.models.fields.CharField', [], {'max_length': '255', 'null': 'True'})
},
'lti_provider.outcomeservice': {
'Meta': {'object_name': 'OutcomeService'},
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'lis_outcome_service_url': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '255'}),
'lti_consumer': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['lti_provider.LtiConsumer']"})
}
}
complete_apps = ['lti_provider']
lms/djangoapps/lti_provider/models.py
......@@ -8,10 +8,13 @@ changes. To do that,
1. Go to the edx-platform dir
2. ./manage.py lms schemamigration lti_provider --auto "description" --settings=devstack
"""
from django.contrib.auth.models import User
from django.db import models
from django.dispatch import receiver
import logging
from courseware.models import SCORE_CHANGED
from xmodule_django.models import CourseKeyField, UsageKeyField
log = logging.getLogger("edx.lti_provider")
class LtiConsumer(models.Model):
......@@ -20,30 +23,99 @@ class LtiConsumer(models.Model):
specific settings, such as the OAuth key/secret pair and any LTI fields
that must be persisted.
"""
consumer_name = models.CharField(max_length=255)
consumer_name = models.CharField(max_length=255, unique=True)
consumer_key = models.CharField(max_length=32, unique=True, db_index=True)
consumer_secret = models.CharField(max_length=32, unique=True)
instance_guid = models.CharField(max_length=255, null=True, unique=True)
@staticmethod
def get_or_supplement(instance_guid, consumer_key):
"""
The instance_guid is the best way to uniquely identify an LTI consumer.
However according to the LTI spec, the instance_guid field is optional
and so cannot be relied upon to be present.
This method first attempts to find an LtiConsumer by instance_guid.
Failing that, it tries to find a record with a matching consumer_key.
This can be the case if the LtiConsumer record was created as the result
of an LTI launch with no instance_guid.
If the instance_guid is now present, the LtiConsumer model will be
supplemented with the instance_guid, to more concretely identify the
consumer.
In practice, nearly all major LTI consumers provide an instance_guid, so
the fallback mechanism of matching by consumer key should be rarely
required.
"""
consumer = None
if instance_guid:
try:
consumer = LtiConsumer.objects.get(instance_guid=instance_guid)
except LtiConsumer.DoesNotExist:
# The consumer may not exist, or its record may not have a guid
pass
# Search by consumer key instead of instance_guid. If there is no
# consumer with a matching key, the LTI launch does not have permission
# to access the content.
if not consumer:
consumer = LtiConsumer.objects.get(
consumer_key=consumer_key,
instance_guid=instance_guid if instance_guid else None
)
# Add the instance_guid field to the model if it's not there already.
if instance_guid and not consumer.instance_guid:
consumer.instance_guid = instance_guid
consumer.save()
return consumer
@receiver(SCORE_CHANGED)
def score_changed_handler(sender, **kwargs): # pylint: disable=unused-argument
class OutcomeService(models.Model):
"""
Consume signals that indicate score changes.
Model for a single outcome service associated with an LTI consumer. Note
that a given consumer may have more than one outcome service URL over its
lifetime, so we need to store the outcome service separately from the
LtiConsumer model.
TODO: This function is a placeholder for integration with the LTI 1.1
outcome service, which will follow in a separate change.
An outcome service can be identified in two ways, depending on the
information provided by an LTI launch. The ideal way to identify the service
is by instance_guid, which should uniquely identify a consumer. However that
field is optional in the LTI launch, and so if it is missing we can fall
back on the consumer key (which should be created uniquely for each consumer
although we don't have a technical way to guarantee that).
Some LTI-specified fields use the prefix lis_; this refers to the IMS
Learning Information Services standard from which LTI inherits some
properties
"""
lis_outcome_service_url = models.CharField(max_length=255, unique=True)
lti_consumer = models.ForeignKey(LtiConsumer)
class GradedAssignment(models.Model):
"""
message = """LTI Provider got score change event:
points_possible: {}
points_earned: {}
user_id: {}
course_id: {}
usage_id: {}
Model representing a single launch of a graded assignment by an individual
user. There will be a row created here only if the LTI consumer may require
a result to be returned from the LTI launch (determined by the presence of
the lis_result_sourcedid parameter in the launch POST). There will be only
one row created for a given usage/consumer combination; repeated launches of
the same content by the same user from the same LTI consumer will not add
new rows to the table.
Some LTI-specified fields use the prefix lis_; this refers to the IMS
Learning Information Services standard from which LTI inherits some
properties
"""
print message.format(
kwargs.get('points_possible', None),
kwargs.get('points_earned', None),
kwargs.get('user_id', None),
kwargs.get('course_id', None),
kwargs.get('usage_id', None),
)
user = models.ForeignKey(User, db_index=True)
course_key = CourseKeyField(max_length=255, db_index=True)
usage_key = UsageKeyField(max_length=255, db_index=True)
outcome_service = models.ForeignKey(OutcomeService)
lis_result_sourcedid = models.CharField(max_length=255, db_index=True)
class Meta(object):
"""
Uniqueness constraints.
"""
unique_together = ('outcome_service', 'lis_result_sourcedid')
lms/djangoapps/lti_provider/outcomes.py 0 → 100644
"""
Helper functions for managing interactions with the LTI outcomes service defined
in LTI v1.1.
"""
import logging
from lxml import etree
from lxml.builder import ElementMaker
import requests
import requests_oauthlib
import uuid
from lti_provider.models import GradedAssignment, OutcomeService
log = logging.getLogger("edx.lti_provider")
def store_outcome_parameters(request_params, user, lti_consumer):
"""
Determine whether a set of LTI launch parameters contains information about
an expected score, and if so create a GradedAssignment record. Create a new
OutcomeService record if none exists for the tool consumer, and update any
incomplete record with additional data if it is available.
"""
result_id = request_params.get('lis_result_sourcedid', None)
# We're only interested in requests that include a lis_result_sourcedid
# parameter. An LTI consumer that does not send that parameter does not
# expect scoring updates for that particular request.
if result_id:
result_service = request_params.get('lis_outcome_service_url', None)
if not result_service:
# TODO: There may be a way to recover from this error; if we know
# the LTI consumer that the request comes from then we may be able
# to figure out the result service URL. As it stands, though, this
# is a badly-formed LTI request
log.warn(
"Outcome Service: lis_outcome_service_url parameter missing "
"from scored assignment; we will be unable to return a score. "
"Request parameters: %s",
request_params
)
return
# Both usage and course ID parameters are supplied in the LTI launch URL
usage_key = request_params['usage_key']
course_key = request_params['course_key']
# Create a record of the outcome service if necessary
outcomes, __ = OutcomeService.objects.get_or_create(
lis_outcome_service_url=result_service,
lti_consumer=lti_consumer
)
GradedAssignment.objects.get_or_create(
lis_result_sourcedid=result_id,
course_key=course_key,
usage_key=usage_key,
user=user,
outcome_service=outcomes
)
def generate_replace_result_xml(result_sourcedid, score):
"""
Create the XML document that contains the new score to be sent to the LTI
consumer. The format of this message is defined in the LTI 1.1 spec.
"""
# Pylint doesn't recognize members in the LXML module
# pylint: disable=no-member
elem = ElementMaker(nsmap={None: 'http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0'})
xml = elem.imsx_POXEnvelopeRequest(
elem.imsx_POXHeader(
elem.imsx_POXRequestHeaderInfo(
elem.imsx_version('V1.0'),
elem.imsx_messageIdentifier(str(uuid.uuid4()))
)
),
elem.imsx_POXBody(
elem.replaceResultRequest(
elem.resultRecord(
elem.sourcedGUID(
elem.sourcedId(result_sourcedid)
),
elem.result(
elem.resultScore(
elem.language('en'),
elem.textString(str(score))
)
)
)
)
)
)
return etree.tostring(xml, xml_declaration=True, encoding='UTF-8')
def sign_and_send_replace_result(assignment, xml):
"""
Take the XML document generated in generate_replace_result_xml, and sign it
with the consumer key and secret assigned to the consumer. Send the signed
message to the LTI consumer.
"""
outcome_service = assignment.outcome_service
consumer = outcome_service.lti_consumer
consumer_key = consumer.consumer_key
consumer_secret = consumer.consumer_secret
# Calculate the OAuth signature for the replace_result message.
# TODO: According to the LTI spec, there should be an additional
# oauth_body_hash field that contains a digest of the replace_result
# message. Testing with Canvas throws an error when this field is included.
# This code may need to be revisited once we test with other LMS platforms,
# and confirm whether there's a bug in Canvas.
oauth = requests_oauthlib.OAuth1(consumer_key, consumer_secret)
headers = {'content-type': 'application/xml'}
response = requests.post(
assignment.outcome_service.lis_outcome_service_url,
data=xml,
auth=oauth,
headers=headers
)
return response
def check_replace_result_response(response):
"""
Parse the response sent by the LTI consumer after an score update message
has been processed. Return True if the message was properly received, or
False if not. The format of this message is defined in the LTI 1.1 spec.
"""
# Pylint doesn't recognize members in the LXML module
# pylint: disable=no-member
if response.status_code != 200:
log.error(
"Outcome service response: Unexpected status code %s",
response.status_code
)
return False
try:
xml = response.content
root = etree.fromstring(xml)
except etree.ParseError as ex:
log.error("Outcome service response: Failed to parse XML: %s\n %s", ex, xml)
return False
major_codes = root.xpath(
'//ns:imsx_codeMajor',
namespaces={'ns': 'http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0'})
if len(major_codes) != 1:
log.error(
"Outcome service response: Expected exactly one imsx_codeMajor field in response. Received %s",
major_codes
)
return False
if major_codes[0].text != 'success':
log.error(
"Outcome service response: Unexpected major code: %s.",
major_codes[0].text
)
return False
return True
lms/djangoapps/lti_provider/signature_validator.py
......@@ -2,8 +2,6 @@
Subclass of oauthlib's RequestValidator that checks an OAuth signature.
"""
from django.core.exceptions import ObjectDoesNotExist
from oauthlib.oauth1 import SignatureOnlyEndpoint
from oauthlib.oauth1 import RequestValidator
......@@ -91,7 +89,7 @@ class SignatureValidator(RequestValidator):
"""
try:
return LtiConsumer.objects.get(consumer_key=client_key).consumer_secret
except ObjectDoesNotExist:
except LtiConsumer.DoesNotExist:
return None
def verify(self, request):
......
lms/djangoapps/lti_provider/tasks.py 0 → 100644
"""
Asynchronous tasks for the LTI provider app.
"""
from django.dispatch import receiver
import logging
from requests.exceptions import RequestException
from courseware.models import SCORE_CHANGED
from lms import CELERY_APP
from lti_provider.models import GradedAssignment
import lti_provider.outcomes
from lti_provider.views import parse_course_and_usage_keys
log = logging.getLogger("edx.lti_provider")
@receiver(SCORE_CHANGED)
def score_changed_handler(sender, **kwargs): # pylint: disable=unused-argument
"""
Consume signals that indicate score changes. See the definition of
courseware.models.SCORE_CHANGED for a description of the signal.
"""
points_possible = kwargs.get('points_possible', None)
points_earned = kwargs.get('points_earned', None)
user_id = kwargs.get('user_id', None)
course_id = kwargs.get('course_id', None)
usage_id = kwargs.get('usage_id', None)
if None not in (points_earned, points_possible, user_id, course_id, user_id):
send_outcome.delay(
points_possible,
points_earned,
user_id,
course_id,
usage_id
)
else:
log.error(
"Outcome Service: Required signal parameter is None. "
"points_possible: %s, points_earned: %s, user_id: %s, "
"course_id: %s, usage_id: %s",
points_possible, points_earned, user_id, course_id, usage_id
)
@CELERY_APP.task
def send_outcome(points_possible, points_earned, user_id, course_id, usage_id):
"""
Calculate the score for a given user in a problem and send it to the
appropriate LTI consumer's outcome service.
"""
course_key, usage_key = parse_course_and_usage_keys(course_id, usage_id)
assignments = GradedAssignment.objects.filter(
user=user_id, course_key=course_key, usage_key=usage_key
)
# Calculate the user's score, on a scale of 0.0 - 1.0.
score = float(points_earned) / float(points_possible)
# There may be zero or more assignment records. We would expect for there
# to be zero if the user/course/usage combination does not relate to a
# previous graded LTI launch. This can happen if an LTI consumer embeds some
# gradable content in a context that doesn't require a score (maybe by
# including an exercise as a sample that students may complete but don't
# count towards their grade).
# There could be more than one GradedAssignment record if the same content
# is embedded more than once in a single course. This would be a strange
# course design on the consumer's part, but we handle it by sending update
# messages for all launches of the content.
for assignment in assignments:
xml = lti_provider.outcomes.generate_replace_result_xml(
assignment.lis_result_sourcedid, score
)
try:
response = lti_provider.outcomes.sign_and_send_replace_result(assignment, xml)
except RequestException:
# failed to send result. 'response' is None, so more detail will be
# logged at the end of the method.
response = None
log.exception("Outcome Service: Error when sending result.")
# If something went wrong, make sure that we have a complete log record.
# That way we can manually fix things up on the campus system later if
# necessary.
if not (response and lti_provider.outcomes.check_replace_result_response(response)):
log.error(
"Outcome Service: Failed to update score on LTI consumer. "
"User: %s, course: %s, usage: %s, score: %s, possible: %s "
"status: %s, body: %s",
user_id,
course_key,
usage_key,
points_earned,
points_possible,
response,
response.text if response else 'Unknown'
)
lms/djangoapps/lti_provider/tests/test_views.py
......@@ -6,9 +6,9 @@ from django.test import TestCase
from django.test.client import RequestFactory
from mock import patch, MagicMock
from lti_provider import views
from lti_provider import views, models
from lti_provider.signature_validator import SignatureValidator
from opaque_keys.edx.keys import CourseKey, UsageKey
from opaque_keys.edx.locator import CourseLocator, BlockUsageLocator
from student.tests.factories import UserFactory
......@@ -23,8 +23,15 @@ LTI_DEFAULT_PARAMS = {
'oauth_nonce': u'OAuth Nonce',
}
COURSE_KEY = CourseKey.from_string('some/course/id')
USAGE_KEY = UsageKey.from_string('i4x://some/course/problem/uuid').map_into_course(COURSE_KEY)
LTI_OPTIONAL_PARAMS = {
'lis_result_sourcedid': u'result sourcedid',
'lis_outcome_service_url': u'outcome service URL',
'tool_consumer_instance_guid': u'consumer instance guid'
}
COURSE_KEY = CourseLocator(org='some_org', course='some_course', run='some_run')
USAGE_KEY = BlockUsageLocator(course_key=COURSE_KEY, block_type='problem', block_id='block_id')
COURSE_PARAMS = {
'course_key': COURSE_KEY,
'usage_key': USAGE_KEY
......@@ -67,6 +74,12 @@ class LtiLaunchTest(TestCase):
super(LtiLaunchTest, self).setUp()
# Always accept the OAuth signature
SignatureValidator.verify = MagicMock(return_value=True)
self.consumer = models.LtiConsumer(
consumer_name='consumer',
consumer_key=LTI_DEFAULT_PARAMS['oauth_consumer_key'],
consumer_secret='secret'
)
self.consumer.save()
@patch('lti_provider.views.render_courseware')
def test_valid_launch(self, render):
......@@ -77,6 +90,20 @@ class LtiLaunchTest(TestCase):
views.lti_launch(request, unicode(COURSE_KEY), unicode(USAGE_KEY))
render.assert_called_with(request, ALL_PARAMS)
@patch('lti_provider.views.render_courseware')
@patch('lti_provider.views.store_outcome_parameters')
def test_outcome_service_registered(self, store_params, _render):
"""
Verifies that the LTI launch succeeds when passed a valid request.
"""
request = build_launch_request()
views.lti_launch(
request,
unicode(COURSE_PARAMS['course_key']),
unicode(COURSE_PARAMS['usage_key'])
)
store_params.assert_called_with(ALL_PARAMS, request.user, self.consumer)
def launch_with_missing_parameter(self, missing_param):
"""
Helper method to remove a parameter from the LTI launch and call the view
......@@ -121,6 +148,33 @@ class LtiLaunchTest(TestCase):
for key in views.REQUIRED_PARAMETERS:
self.assertEqual(session[key], request.POST[key], key + ' not set in the session')
@patch('lti_provider.views.lti_run')
def test_optional_parameters_in_session(self, _run):
"""
Verifies that the outcome-related optional LTI parameters are properly
stored in the session
"""
request = build_launch_request()
request.POST.update(LTI_OPTIONAL_PARAMS)
views.lti_launch(
request,
unicode(COURSE_PARAMS['course_key']),
unicode(COURSE_PARAMS['usage_key'])
)
session = request.session[views.LTI_SESSION_KEY]
self.assertEqual(
session['lis_result_sourcedid'], u'result sourcedid',
'Result sourcedid not set in the session'
)
self.assertEqual(
session['lis_outcome_service_url'], u'outcome service URL',
'Outcome service URL not set in the session'
)
self.assertEqual(
session['tool_consumer_instance_guid'], u'consumer instance guid',
'Consumer instance GUID not set in the session'
)
def test_redirect_for_non_authenticated_user(self):
"""
Verifies that if the lti_launch view is called by an unauthenticated
......@@ -149,6 +203,15 @@ class LtiRunTest(TestCase):
Tests for the lti_run view
"""
def setUp(self):
super(LtiRunTest, self).setUp()
consumer = models.LtiConsumer(
consumer_name='consumer',
consumer_key=LTI_DEFAULT_PARAMS['oauth_consumer_key'],
consumer_secret='secret'
)
consumer.save()
@patch('lti_provider.views.render_courseware')
def test_valid_launch(self, render):
"""
......
lms/djangoapps/lti_provider/views.py
......@@ -14,6 +14,8 @@ from courseware.access import has_access
from courseware.courses import get_course_with_access
from courseware.module_render import get_module_by_usage_id
from edxmako.shortcuts import render_to_response
from lti_provider.outcomes import store_outcome_parameters
from lti_provider.models import LtiConsumer
from lti_provider.signature_validator import SignatureValidator
from lms_xblock.runtime import unquote_slashes
from opaque_keys.edx.keys import CourseKey, UsageKey
......@@ -29,6 +31,11 @@ REQUIRED_PARAMETERS = [
'oauth_nonce'
]
OPTIONAL_PARAMETERS = [
'lis_result_sourcedid', 'lis_outcome_service_url',
'tool_consumer_instance_guid'
]
LTI_SESSION_KEY = 'lti_provider_parameters'
......@@ -61,12 +68,17 @@ def lti_launch(request, course_id, usage_id):
return HttpResponseForbidden()
# Check the OAuth signature on the message
if not SignatureValidator().verify(request):
try:
if not SignatureValidator().verify(request):
return HttpResponseForbidden()
except LtiConsumer.DoesNotExist:
return HttpResponseForbidden()
params = get_required_parameters(request.POST)
if not params:
return HttpResponseBadRequest()
params.update(get_optional_parameters(request.POST))
# Store the course, and usage ID in the session to prevent privilege
# escalation if a staff member in one course tries to access material in
# another.
......@@ -118,6 +130,15 @@ def lti_run(request):
# Remove the parameters from the session to prevent replay
del request.session[LTI_SESSION_KEY]
# Store any parameters required by the outcome service in order to report
# scores back later. We know that the consumer exists, since the record was
# used earlier to verify the oauth signature.
lti_consumer = LtiConsumer.get_or_supplement(
params.get('tool_consumer_instance_guid', None),
params['oauth_consumer_key']
)
store_outcome_parameters(params, request.user, lti_consumer)
return render_courseware(request, params)
......@@ -143,6 +164,19 @@ def get_required_parameters(dictionary, additional_params=None):
return params
def get_optional_parameters(dictionary):
"""
Extract all optional LTI parameters from a dictionary. This method does not
fail if any parameters are missing.
:param dictionary: A dictionary containing zero or more optional parameters.
:return: A new dictionary containing all optional parameters from the
original dictionary, or an empty dictionary if no optional parameters
were present.
"""
return {key: dictionary[key] for key in OPTIONAL_PARAMETERS if key in dictionary}
def restore_params_from_session(request):
"""
Fetch the parameters that were stored in the session by an LTI launch, and
......@@ -157,7 +191,10 @@ def restore_params_from_session(request):
return None
session_params = request.session[LTI_SESSION_KEY]
additional_params = ['course_key', 'usage_key']
return get_required_parameters(session_params, additional_params)
for key in REQUIRED_PARAMETERS + additional_params:
if key not in session_params:
return None
return session_params
def render_courseware(request, lti_params):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment