Skip to content

E
edx-platform
Commit 1d517bd0 by Oleg Marshev
Options

Merge pull request #2029 from edx/oleg/lti_stub_refactor 

Refactor stub implementation of LTI Provider. BLD-601.
parents 15fc640e 79ffcb7e
Showing with 335 additions and 561 deletions
CHANGELOG.rst
......@@ -5,6 +5,8 @@ These are notable changes in edx-platform. This is a rolling list of changes,
in roughly chronological order, most recent first. Add your entries at or near
the top. Include a label indicating the component affected.
Blades: Refactor stub implementation of LTI Provider. BLD-601.
LMS: In left accordion and progress page, due dates are now displayed in time
zone specified by settings.TIME_ZONE, instead of UTC always
......
common/djangoapps/terrain/start_stubs.py
......@@ -6,11 +6,12 @@ from lettuce import before, after, world
from django.conf import settings
from terrain.stubs.youtube import StubYouTubeService
from terrain.stubs.xqueue import StubXQueueService
from terrain.stubs.lti import StubLtiService
SERVICES = {
"youtube": {"port": settings.YOUTUBE_PORT, "class": StubYouTubeService},
"xqueue": {"port": settings.XQUEUE_PORT, "class": StubXQueueService},
"lti": {"port": settings.LTI_PORT, "class": StubLtiService},
}
......
common/djangoapps/terrain/stubs/http.py
......@@ -236,7 +236,7 @@ class StubHttpService(HTTPServer, object):
Configure the server to listen on localhost.
Default is to choose an arbitrary open port.
"""
address = ('127.0.0.1', port_num)
address = ('0.0.0.0', port_num)
HTTPServer.__init__(self, address, self.HANDLER_CLASS)
# Create a dict to store configuration values set by the client
......
common/djangoapps/terrain/stubs/lti.py 0 → 100644
"""
Stub implementation of LTI Provider.
What is supported:
------------------
1.) This LTI Provider can service only one Tool Consumer at the same time. It is
not possible to have this LTI multiple times on a single page in LMS.
"""
from uuid import uuid4
import textwrap
import urllib
import re
from oauthlib.oauth1.rfc5849 import signature
import oauthlib.oauth1
import hashlib
import base64
import mock
import requests
from http import StubHttpRequestHandler, StubHttpService
class StubLtiHandler(StubHttpRequestHandler):
"""
A handler for LTI POST and GET requests.
"""
DEFAULT_CLIENT_KEY = 'test_client_key'
DEFAULT_CLIENT_SECRET = 'test_client_secret'
DEFAULT_LTI_ENDPOINT = 'correct_lti_endpoint'
DEFAULT_LTI_ADDRESS = 'http://127.0.0.1:{port}/'
def do_GET(self):
"""
Handle a GET request from the client and sends response back.
Used for checking LTI Provider started correctly.
"""
self.send_response(200, 'This is LTI Provider.', {'Content-type': 'text/plain'})
def do_POST(self):
"""
Handle a POST request from the client and sends response back.
"""
if 'grade' in self.path and self._send_graded_result().status_code == 200:
status_message = 'LTI consumer (edX) responded with XML content:<br>' + self.server.grade_data['TC answer']
content = self._create_content(status_message)
self.send_response(200, content)
# Respond to request with correct lti endpoint
elif self._is_correct_lti_request():
params = {k: v for k, v in self.post_dict.items() if k != 'oauth_signature'}
if self._check_oauth_signature(params, self.post_dict.get('oauth_signature', "")):
status_message = "This is LTI tool. Success."
# Set data for grades what need to be stored as server data
if 'lis_outcome_service_url' in self.post_dict:
self.server.grade_data = {
'callback_url': self.post_dict.get('lis_outcome_service_url'),
'sourcedId': self.post_dict.get('lis_result_sourcedid')
}
submit_url = '//{}:{}'.format(*self.server.server_address)
content = self._create_content(status_message, submit_url)
self.send_response(200, content)
else:
content = self._create_content("Wrong LTI signature")
self.send_response(200, content)
else:
content = self._create_content("Invalid request URL")
self.send_response(500, content)
def _send_graded_result(self):
"""
Send grade request.
"""
values = {
'textString': 0.5,
'sourcedId': self.server.grade_data['sourcedId'],
'imsx_messageIdentifier': uuid4().hex,
}
payload = textwrap.dedent("""
<?xml version = "1.0" encoding = "UTF-8"?>
<imsx_POXEnvelopeRequest xmlns="http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0">
<imsx_POXHeader>
<imsx_POXRequestHeaderInfo>
<imsx_version>V1.0</imsx_version>
<imsx_messageIdentifier>{imsx_messageIdentifier}</imsx_messageIdentifier> /
</imsx_POXRequestHeaderInfo>
</imsx_POXHeader>
<imsx_POXBody>
<replaceResultRequest>
<resultRecord>
<sourcedGUID>
<sourcedId>{sourcedId}</sourcedId>
</sourcedGUID>
<result>
<resultScore>
<language>en-us</language>
<textString>{textString}</textString>
</resultScore>
</result>
</resultRecord>
</replaceResultRequest>
</imsx_POXBody>
</imsx_POXEnvelopeRequest>
""")
data = payload.format(**values)
url = self.server.grade_data['callback_url']
headers = {
'Content-Type': 'application/xml',
'X-Requested-With': 'XMLHttpRequest',
'Authorization': self._oauth_sign(url, data)
}
# Send request ignoring verifirecation of SSL certificate
response = requests.post(url, data=data, headers=headers, verify=False)
self.server.grade_data['TC answer'] = response.content
return response
def _create_content(self, response_text, submit_url=None):
"""
Return content (str) either for launch, send grade or get result from TC.
"""
if submit_url:
submit_form = textwrap.dedent("""
<form action="{}/grade" method="post">
<input type="submit" name="submit-button" value="Submit">
</form>
""").format(submit_url)
else:
submit_form = ''
# Show roles only for LTI launch.
if self.post_dict.get('roles'):
role = '<h5>Role: {}</h5>'.format(self.post_dict['roles'])
else:
role = ''
response_str = textwrap.dedent("""
<html>
<head>
<title>TEST TITLE</title>
</head>
<body>
<div>
<h2>IFrame loaded</h2>
<h3>Server response is:</h3>
<h3 class="result">{response}</h3>
{role}
</div>
{submit_form}
</body>
</html>
""").format(response=response_text, role=role, submit_form=submit_form)
# Currently LTI module doublequotes the lis_result_sourcedid parameter.
# Unquote response two times.
return urllib.unquote(urllib.unquote(response_str))
def _is_correct_lti_request(self):
"""
Return a boolean indicating whether the URL path is a valid LTI end-point.
"""
lti_endpoint = self.server.config.get('lti_endpoint', self.DEFAULT_LTI_ENDPOINT)
return lti_endpoint in self.path
def _oauth_sign(self, url, body):
"""
Signs request and returns signed body and headers.
"""
client_key = self.server.config.get('client_key', self.DEFAULT_CLIENT_KEY)
client_secret = self.server.config.get('client_secret', self.DEFAULT_CLIENT_SECRET)
client = oauthlib.oauth1.Client(
client_key=unicode(client_key),
client_secret=unicode(client_secret)
)
headers = {
# This is needed for body encoding:
'Content-Type': 'application/x-www-form-urlencoded',
}
# Calculate and encode body hash. See http://oauth.googlecode.com/svn/spec/ext/body_hash/1.0/oauth-bodyhash.html
sha1 = hashlib.sha1()
sha1.update(body)
oauth_body_hash = base64.b64encode(sha1.digest())
__, headers, __ = client.sign(
unicode(url.strip()),
http_method=u'POST',
body={u'oauth_body_hash': oauth_body_hash},
headers=headers
)
headers = headers['Authorization'] + ', oauth_body_hash="{}"'.format(oauth_body_hash)
return headers
def _check_oauth_signature(self, params, client_signature):
"""
Checks oauth signature from client.
`params` are params from post request except signature,
`client_signature` is signature from request.
Builds mocked request and verifies hmac-sha1 signing::
1. builds string to sign from `params`, `url` and `http_method`.
2. signs it with `client_secret` which comes from server settings.
3. obtains signature after sign and then compares it with request.signature
(request signature comes form client in request)
Returns `True` if signatures are correct, otherwise `False`.
"""
client_secret = unicode(self.server.config.get('client_secret', self.DEFAULT_CLIENT_SECRET))
port = self.server.server_address[1]
lti_base = self.DEFAULT_LTI_ADDRESS.format(port=port)
lti_endpoint = self.server.config.get('lti_endpoint', self.DEFAULT_LTI_ENDPOINT)
url = lti_base + lti_endpoint
request = mock.Mock()
request.params = [(unicode(k), unicode(v)) for k, v in params.items()]
request.uri = unicode(url)
request.http_method = u'POST'
request.signature = unicode(client_signature)
return signature.verify_hmac_sha1(request, client_secret)
class StubLtiService(StubHttpService):
"""
A stub LTI provider server that responds
to POST and GET requests to localhost.
"""
HANDLER_CLASS = StubLtiHandler
common/djangoapps/terrain/stubs/start.py
......@@ -8,6 +8,7 @@ from .comments import StubCommentsService
from .xqueue import StubXQueueService
from .youtube import StubYouTubeService
from .ora import StubOraService
from .lti import StubLtiService
USAGE = "USAGE: python -m stubs.start SERVICE_NAME PORT_NUM [CONFIG_KEY=CONFIG_VAL, ...]"
......@@ -17,6 +18,7 @@ SERVICES = {
'youtube': StubYouTubeService,
'ora': StubOraService,
'comments': StubCommentsService,
'lti': StubLtiService,
}
# Log to stdout, including debug messages
......
common/djangoapps/terrain/stubs/tests/test_lti_stub.py 0 → 100644
"""
Unit tests for stub LTI implementation.
"""
from mock import Mock, patch
import unittest
import urllib2
import requests
from terrain.stubs.lti import StubLtiService
class StubLtiServiceTest(unittest.TestCase):
"""
A stub of the LTI provider that listens on a local
port and responds with pre-defined grade messages.
Used for lettuce BDD tests in lms/courseware/features/lti.feature
"""
def setUp(self):
self.server = StubLtiService()
self.uri = 'http://127.0.0.1:{}/'.format(self.server.port)
self.launch_uri = self.uri + 'correct_lti_endpoint'
self.addCleanup(self.server.shutdown)
self.payload = {
'user_id': 'default_user_id',
'roles': 'Student',
'oauth_nonce': '',
'oauth_timestamp': '',
'oauth_consumer_key': 'test_client_key',
'lti_version': 'LTI-1p0',
'oauth_signature_method': 'HMAC-SHA1',
'oauth_version': '1.0',
'oauth_signature': '',
'lti_message_type': 'basic-lti-launch-request',
'oauth_callback': 'about:blank',
'launch_presentation_return_url': '',
'lis_outcome_service_url': 'http://localhost:8001/test_callback',
'lis_result_sourcedid': '',
'resource_link_id':'',
}
def test_invalid_request_url(self):
"""
Tests that LTI server processes request with right program path but with wrong header.
"""
self.launch_uri = self.uri + 'wrong_lti_endpoint'
response = requests.post(self.launch_uri, data=self.payload)
self.assertIn('Invalid request URL', response.content)
def test_wrong_signature(self):
"""
Tests that LTI server processes request with right program
path and responses with incorrect signature.
"""
response = requests.post(self.launch_uri, data=self.payload)
self.assertIn('Wrong LTI signature', response.content)
@patch('terrain.stubs.lti.signature.verify_hmac_sha1', return_value=True)
def test_success_response_launch_lti(self, check_oauth):
"""
Success lti launch.
"""
response = requests.post(self.launch_uri, data=self.payload)
self.assertIn('This is LTI tool. Success.', response.content)
@patch('terrain.stubs.lti.signature.verify_hmac_sha1', return_value=True)
def test_send_graded_result(self, verify_hmac):
response = requests.post(self.launch_uri, data=self.payload)
self.assertIn('This is LTI tool. Success.', response.content)
grade_uri = self.uri + 'grade'
with patch('terrain.stubs.lti.requests.post') as mocked_post:
mocked_post.return_value = Mock(content='Test response', status_code=200)
response = urllib2.urlopen(grade_uri, data='')
self.assertIn('Test response', response.read())
common/lib/xmodule/xmodule/lti_module.py
......@@ -289,7 +289,7 @@ class LTIModule(LTIFields, XModule):
While testing locally and on Jenkins, mock_lti_server use http.referer
to obtain scheme, so it is ok to have http(s) anyway.
"""
scheme = 'http' if 'sandbox' in self.system.hostname else 'https'
scheme = 'http' if 'sandbox' in self.system.hostname or self.system.debug else 'https'
uri = '{scheme}://{host}{path}'.format(
scheme=scheme,
host=self.system.hostname,
......@@ -325,7 +325,11 @@ class LTIModule(LTIFields, XModule):
the link being launched.
lti_id should be context_id by meaning.
"""
return u':'.join(urllib.quote(i) for i in (self.lti_id, self.get_resource_link_id(), self.get_user_id()))
return "{id}:{resource_link}:{user_id}".format(
id=urllib.quote(self.lti_id),
resource_link=urllib.quote(self.get_resource_link_id()),
user_id=urllib.quote(self.get_user_id())
)
def get_course(self):
"""
......
common/lib/xmodule/xmodule/tests/test_lti_unit.py
......@@ -246,7 +246,8 @@ class LTIModuleTest(LogicTest):
self.assertEqual(real_user_id, expected_user_id)
def test_outcome_service_url(self):
expected_outcome_service_url = 'https://{host}{path}'.format(
expected_outcome_service_url = '{scheme}://{host}{path}'.format(
scheme='http' if self.xmodule.runtime.debug else 'https',
host=self.xmodule.runtime.hostname,
path=self.xmodule.runtime.handler_url(self.xmodule, 'grade_handler', thirdparty=True).rstrip('/?')
)
......
lms/djangoapps/courseware/features/lti.py
......@@ -9,6 +9,7 @@ from splinter.exceptions import ElementDoesNotExist
from django.contrib.auth.models import User
from django.core.urlresolvers import reverse
from django.conf import settings
from lettuce import world, step
from lettuce.django import django_url
......@@ -81,10 +82,7 @@ def incorrect_lti_is_rendered(_step):
def set_correct_lti_passport(_step, user='Instructor'):
coursenum = 'test_course'
metadata = {
'lti_passports': ["correct_lti_id:{}:{}".format(
world.lti_server.oauth_settings['client_key'],
world.lti_server.oauth_settings['client_secret']
)]
'lti_passports': ["correct_lti_id:test_client_key:test_client_secret"]
}
i_am_registered_for_the_course(coursenum, metadata, user)
......@@ -94,10 +92,7 @@ def set_correct_lti_passport(_step, user='Instructor'):
def set_incorrect_lti_passport(_step):
coursenum = 'test_course'
metadata = {
'lti_passports': ["test_lti_id:{}:{}".format(
world.lti_server.oauth_settings['client_key'],
"incorrect_lti_secret_key"
)]
'lti_passports': ["test_lti_id:test_client_key:incorrect_lti_secret_key"]
}
i_am_registered_for_the_course(coursenum, metadata)
......@@ -108,7 +103,7 @@ def add_correct_lti_to_course(_step, fields):
category = 'lti'
metadata = {
'lti_id': 'correct_lti_id',
'launch_url': world.lti_server.oauth_settings['lti_base'] + world.lti_server.oauth_settings['lti_endpoint'],
'launch_url': 'http://127.0.0.1:{}/correct_lti_endpoint'.format(settings.LTI_PORT),
}
if fields.strip() == 'incorrect_lti_id': # incorrect fields
......
lms/djangoapps/courseware/features/lti_setup.py deleted 100644 → 0
#pylint: disable=C0111
#pylint: disable=W0621
from courseware.mock_lti_server.mock_lti_server import MockLTIServer
from lettuce import before, after, world
from django.conf import settings
import threading
from logging import getLogger
logger = getLogger(__name__)
@before.all
def setup_mock_lti_server():
server_host = '127.0.0.1'
server_port = settings.LTI_PORT
address = (server_host, server_port)
# Create the mock server instance
server = MockLTIServer(address)
logger.debug("LTI server started at {} port".format(str(server_port)))
# Start the server running in a separate daemon thread
# Because the thread is a daemon, it will terminate
# when the main thread terminates.
server_thread = threading.Thread(target=server.serve_forever)
server_thread.daemon = True
server_thread.start()
server.server_host = server_host
server.oauth_settings = {
'client_key': 'test_client_key',
'client_secret': 'test_client_secret',
'lti_base': 'http://{}:{}/'.format(server_host, server_port),
'lti_endpoint': 'correct_lti_endpoint'
}
# For testing on localhost make callback url using referer host.
server.real_callback_url_on = False
# Store the server instance in lettuce's world
# so that other steps can access it
# (and we can shut it down later)
world.lti_server = server
@after.all
def teardown_mock_lti_server(total):
# Stop the LTI server and free up the port
world.lti_server.shutdown()
lms/djangoapps/courseware/mock_lti_server/mock_lti_server.py deleted 100644 → 0
"""
LTI Server
What is supported:
------------------
1.) This LTI Provider can service only one Tool Consumer at the same time. It is
not possible to have this LTI multiple times on a single page in LMS.
"""
from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
from uuid import uuid4
import textwrap
import urlparse
from oauthlib.oauth1.rfc5849 import signature
import oauthlib.oauth1
import hashlib
import base64
import mock
import sys
import requests
import textwrap
from logging import getLogger
logger = getLogger(__name__)
class MockLTIRequestHandler(BaseHTTPRequestHandler):
'''
A handler for LTI POST requests.
'''
protocol = "HTTP/1.0"
callback_url = None
def log_message(self, format, *args):
"""Log an arbitrary message."""
# Code copied from BaseHTTPServer.py. Changed to write to sys.stdout
# so that messages won't pollute test output.
sys.stdout.write("%s - - [%s] %s\n" %
(self.client_address[0],
self.log_date_time_string(),
format % args))
def do_GET(self):
'''
Handle a GET request from the client and sends response back.
Used for checking LTI Provider started correctly.
'''
self.send_response(200, 'OK')
self.send_header('Content-type', 'html')
self.end_headers()
response_str = """<html><head><title>TEST TITLE</title></head>
<body>This is LTI Provider.</body></html>"""
self.wfile.write(response_str)
def do_POST(self):
'''
Handle a POST request from the client and sends response back.
'''
if 'grade' in self.path and self._send_graded_result().status_code == 200:
status_message = 'LTI consumer (edX) responded with XML content:<br>' + self.server.grade_data['TC answer']
self.server.grade_data = None
self._send_response(status_message, 200)
# Respond to request with correct lti endpoint:
elif self._is_correct_lti_request():
self.post_dict = self._post_dict()
params = {k: v for k, v in self.post_dict.items() if k != 'oauth_signature'}
if self.server.check_oauth_signature(params, self.post_dict.get('oauth_signature', "")):
status_message = "This is LTI tool. Success."
# set data for grades what need to be stored as server data
if 'lis_outcome_service_url' in self.post_dict:
self.server.grade_data = {
'callback_url': self.post_dict.get('lis_outcome_service_url'),
'sourcedId': self.post_dict.get('lis_result_sourcedid')
}
else:
status_message = "Wrong LTI signature"
self._send_response(status_message, 200)
else:
status_message = "Invalid request URL"
self._send_response(status_message, 500)
def _send_head(self, status_code):
'''
Send the response code and MIME headers
'''
self.send_response(status_code)
self.send_header('Content-type', 'text/html')
self.end_headers()
def _post_dict(self):
'''
Retrieve the POST parameters from the client as a dictionary
'''
try:
length = int(self.headers.getheader('content-length'))
post_dict = urlparse.parse_qs(self.rfile.read(length), keep_blank_values=True)
# The POST dict will contain a list of values for each key.
# None of our parameters are lists, however, so we map [val] --> val.
# If the list contains multiple entries, we pick the first one
post_dict = {key: val[0] for key, val in post_dict.items()}
except:
# We return an empty dict here, on the assumption
# that when we later check that the request has
# the correct fields, it won't find them,
# and will therefore send an error response
return {}
try:
cookie = self.headers.getheader('cookie')
self.server.cookie = {k.strip(): v[0] for k, v in urlparse.parse_qs(cookie).items()}
except:
self.server.cookie = {}
referer = urlparse.urlparse(self.headers.getheader('referer'))
self.server.referer_host = "{}://{}".format(referer.scheme, referer.netloc)
return post_dict
def _send_graded_result(self):
"""
Send grade request.
"""
values = {
'textString': 0.5,
'sourcedId': self.server.grade_data['sourcedId'],
'imsx_messageIdentifier': uuid4().hex,
}
payload = textwrap.dedent("""
<?xml version = "1.0" encoding = "UTF-8"?>
<imsx_POXEnvelopeRequest xmlns="http://www.imsglobal.org/services/ltiv1p1/xsd/imsoms_v1p0">
<imsx_POXHeader>
<imsx_POXRequestHeaderInfo>
<imsx_version>V1.0</imsx_version>
<imsx_messageIdentifier>{imsx_messageIdentifier}</imsx_messageIdentifier> /
</imsx_POXRequestHeaderInfo>
</imsx_POXHeader>
<imsx_POXBody>
<replaceResultRequest>
<resultRecord>
<sourcedGUID>
<sourcedId>{sourcedId}</sourcedId>
</sourcedGUID>
<result>
<resultScore>
<language>en-us</language>
<textString>{textString}</textString>
</resultScore>
</result>
</resultRecord>
</replaceResultRequest>
</imsx_POXBody>
</imsx_POXEnvelopeRequest>
""")
data = payload.format(**values)
if getattr(self.server, 'use_real_callback_url', None):
# Use exact URL that was sent from TC when using this Stub LTI server
# as TP in real standalone environment.
url = self.server.grade_data['callback_url']
else:
# Use relative URL when using TP locally for manual testing or jenkins.
relative_url = urlparse.urlparse(self.server.grade_data['callback_url']).path
url = self.server.referer_host + relative_url
headers = {'Content-Type': 'application/xml', 'X-Requested-With': 'XMLHttpRequest'}
headers['Authorization'] = self.oauth_sign(url, data)
# We can't mock requests in unit tests, because we use them, but we need
# them to be mocked only for this one case.
if getattr(self.server, 'run_inside_unittest_flag', None):
response = mock.Mock(status_code=200, url=url, data=data, headers=headers)
return response
# Send request ignoring verification of SSL certificate
response = requests.post(
url,
data=data,
headers=headers,
verify=False
)
self.server.grade_data['TC answer'] = response.content
return response
def _send_response(self, message, status_code):
'''
Send message back to the client
'''
self._send_head(status_code)
if getattr(self.server, 'grade_data', False): # lti can be graded
url = "//%s:%s" % self.server.server_address
response_str = textwrap.dedent("""
<html>
<head>
<title>TEST TITLE</title>
</head>
<body>
<div>
<h2>Graded IFrame loaded</h2>
<h3>Server response is:</h3>
<h3 class="result">{}</h3>
<h5>Role: {role}</h5>
</div>
<form action="{url}/grade" method="post">
<input type="submit" name="submit-button" value="Submit">
</form>
</body>
</html>
""").format(message, role=self.post_dict['roles'], url=url)
else: # lti can't be graded
response_str = textwrap.dedent("""
<html>
<head>
<title>TEST TITLE</title>
</head>
<body>
<div>
<h2>IFrame loaded</h2>
<h3>Server response is:</h3>
<h3 class="result">{}</h3>
</div>
</body>
</html>
""").format(message)
logger.debug("LTI: sent response {}".format(response_str))
self.wfile.write(response_str)
def _is_correct_lti_request(self):
'''
If url to LTI tool is correct.
'''
return self.server.oauth_settings['lti_endpoint'] in self.path
def oauth_sign(self, url, body):
"""
Signs request and returns signed body and headers.
"""
client = oauthlib.oauth1.Client(
client_key=unicode(self.server.oauth_settings['client_key']),
client_secret=unicode(self.server.oauth_settings['client_secret'])
)
headers = {
# This is needed for body encoding:
'Content-Type': 'application/x-www-form-urlencoded',
}
#Calculate and encode body hash. See http://oauth.googlecode.com/svn/spec/ext/body_hash/1.0/oauth-bodyhash.html
sha1 = hashlib.sha1()
sha1.update(body)
oauth_body_hash = base64.b64encode(sha1.digest())
__, headers, __ = client.sign(
unicode(url.strip()),
http_method=u'POST',
body={u'oauth_body_hash': oauth_body_hash},
headers=headers
)
headers = headers['Authorization'] + ', oauth_body_hash="{}"'.format(oauth_body_hash)
return headers
class MockLTIServer(HTTPServer):
'''
A mock LTI provider server that responds
to POST requests to localhost.
'''
def __init__(self, address):
'''
Initialize the mock XQueue server instance.
*address* is the (host, host's port to listen to) tuple.
'''
handler = MockLTIRequestHandler
HTTPServer.__init__(self, address, handler)
def shutdown(self):
'''
Stop the server and free up the port
'''
# First call superclass shutdown()
HTTPServer.shutdown(self)
# We also need to manually close the socket
self.socket.close()
def check_oauth_signature(self, params, client_signature):
'''
Checks oauth signature from client.
`params` are params from post request except signature,
`client_signature` is signature from request.
Builds mocked request and verifies hmac-sha1 signing::
1. builds string to sign from `params`, `url` and `http_method`.
2. signs it with `client_secret` which comes from server settings.
3. obtains signature after sign and then compares it with request.signature
(request signature comes form client in request)
Returns `True` if signatures are correct, otherwise `False`.
'''
client_secret = unicode(self.oauth_settings['client_secret'])
url = self.oauth_settings['lti_base'] + self.oauth_settings['lti_endpoint']
request = mock.Mock()
request.params = [(unicode(k), unicode(v)) for k, v in params.items()]
request.uri = unicode(url)
request.http_method = u'POST'
request.signature = unicode(client_signature)
return signature.verify_hmac_sha1(request, client_secret)
lms/djangoapps/courseware/mock_lti_server/server_start.py deleted 100644 → 0
"""
Mock LTI server for manual testing.
Used for manual testing and testing on sandbox.
"""
from mock_lti_server import MockLTIServer
server_port = 8034
server_host = 'localhost'
address = (server_host, server_port)
server = MockLTIServer(address)
server.oauth_settings = {
'client_key': 'test_client_key',
'client_secret': 'test_client_secret',
'lti_base': 'http://{}:{}/'.format(server_host, server_port),
'lti_endpoint': 'correct_lti_endpoint'
}
server.server_host = server_host
server.server_port = server_port
# For testing on localhost make callback url using referer host.
server.use_real_callback_url = False
try:
server.serve_forever()
except KeyboardInterrupt:
print('^C received, shutting down server')
server.socket.close()
lms/djangoapps/courseware/mock_lti_server/test_mock_lti_server.py deleted 100644 → 0
"""
Test for Mock_LTI_Server
"""
from mock import Mock
import unittest
import threading
import requests
from mock_lti_server import MockLTIServer
class MockLTIServerTest(unittest.TestCase):
'''
A mock version of the LTI provider server that listens on a local
port and responds with pre-defined grade messages.
Used for lettuce BDD tests in lms/courseware/features/lti.feature
'''
def setUp(self):
# Create the server
server_port = 8034
server_host = 'localhost'
address = (server_host, server_port)
self.server = MockLTIServer(address)
self.server.oauth_settings = {
'client_key': 'test_client_key',
'client_secret': 'test_client_secret',
'lti_base': 'http://{}:{}/'.format(server_host, server_port),
'lti_endpoint': 'correct_lti_endpoint'
}
self.server.run_inside_unittest_flag = True
#flag for creating right callback_url
self.server.test_mode = True
self.server.server_host = server_host
self.server.server_port = server_port
# Start the server in a separate daemon thread
server_thread = threading.Thread(target=self.server.serve_forever)
server_thread.daemon = True
server_thread.start()
def tearDown(self):
# Stop the server, freeing up the port
self.server.shutdown()
def test_wrong_header(self):
"""
Tests that LTI server processes request with right program path but with wrong header.
"""
#wrong number of params and no signature
payload = {
'user_id': 'default_user_id',
'roles': 'Student',
'oauth_nonce': '',
'oauth_timestamp': '',
}
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
headers = {'referer': 'http://localhost:8000/'}
response = requests.post(uri, data=payload, headers=headers)
self.assertIn('Wrong LTI signature', response.content)
def test_wrong_signature(self):
"""
Tests that LTI server processes request with right program
path and responses with incorrect signature.
"""
payload = {
'user_id': 'default_user_id',
'roles': 'Student',
'oauth_nonce': '',
'oauth_timestamp': '',
'oauth_consumer_key': 'test_client_key',
'lti_version': 'LTI-1p0',
'oauth_signature_method': 'HMAC-SHA1',
'oauth_version': '1.0',
'oauth_signature': '',
'lti_message_type': 'basic-lti-launch-request',
'oauth_callback': 'about:blank',
'launch_presentation_return_url': '',
'lis_outcome_service_url': '',
'lis_result_sourcedid': '',
'resource_link_id':'',
}
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
headers = {'referer': 'http://localhost:8000/'}
response = requests.post(uri, data=payload, headers=headers)
self.assertIn('Wrong LTI signature', response.content)
def test_success_response_launch_lti(self):
"""
Success lti launch.
"""
payload = {
'user_id': 'default_user_id',
'roles': 'Student',
'oauth_nonce': '',
'oauth_timestamp': '',
'oauth_consumer_key': 'test_client_key',
'lti_version': 'LTI-1p0',
'oauth_signature_method': 'HMAC-SHA1',
'oauth_version': '1.0',
'oauth_signature': '',
'lti_message_type': 'basic-lti-launch-request',
'oauth_callback': 'about:blank',
'launch_presentation_return_url': '',
'lis_outcome_service_url': '',
'lis_result_sourcedid': '',
'resource_link_id':'',
}
self.server.check_oauth_signature = Mock(return_value=True)
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
headers = {'referer': 'http://localhost:8000/'}
response = requests.post(uri, data=payload, headers=headers)
self.assertIn('This is LTI tool. Success.', response.content)
def test_send_graded_result(self):
payload = {
'user_id': 'default_user_id',
'roles': 'Student',
'oauth_nonce': '',
'oauth_timestamp': '',
'oauth_consumer_key': 'test_client_key',
'lti_version': 'LTI-1p0',
'oauth_signature_method': 'HMAC-SHA1',
'oauth_version': '1.0',
'oauth_signature': '',
'lti_message_type': 'basic-lti-launch-request',
'oauth_callback': 'about:blank',
'launch_presentation_return_url': '',
'lis_outcome_service_url': '',
'lis_result_sourcedid': '',
'resource_link_id':'',
}
self.server.check_oauth_signature = Mock(return_value=True)
uri = self.server.oauth_settings['lti_base'] + self.server.oauth_settings['lti_endpoint']
#this is the uri for sending grade from lti
headers = {'referer': 'http://localhost:8000/'}
response = requests.post(uri, data=payload, headers=headers)
self.assertIn('This is LTI tool. Success.', response.content)
self.server.grade_data['TC answer'] = "Test response"
graded_response = requests.post('http://127.0.0.1:8034/grade')
self.assertIn('Test response', graded_response.content)
lms/djangoapps/courseware/tests/test_lti_integration.py
......@@ -31,7 +31,11 @@ class TestLTI(BaseTestXmodule):
module_id = unicode(urllib.quote(self.item_module.id))
user_id = unicode(self.item_descriptor.xmodule_runtime.anonymous_student_id)
sourcedId = u':'.join(urllib.quote(i) for i in (lti_id, module_id, user_id))
sourcedId = "{id}:{resource_link}:{user_id}".format(
id=urllib.quote(lti_id),
resource_link=urllib.quote(module_id),
user_id=urllib.quote(user_id)
)
lis_outcome_service_url = 'https://{host}{path}'.format(
host=self.item_descriptor.xmodule_runtime.hostname,
......
lms/envs/acceptance.py
......@@ -13,6 +13,7 @@ from .sauce import *
# You need to start the server in debug mode,
# otherwise the browser will not render the pages correctly
DEBUG = True
SITE_NAME = 'localhost:{}'.format(LETTUCE_SERVER_PORT)
# Output Django logs to a file
import logging
......
lms/envs/devstack.py
......@@ -7,6 +7,7 @@ from .aws import * # pylint: disable=wildcard-import, unused-wildcard-import
DEBUG = True
USE_I18N = True
TEMPLATE_DEBUG = True
SITE_NAME = 'localhost:8000'
# By default don't use a worker, execute tasks as if they were local functions
CELERY_ALWAYS_EAGER = True
......
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