-
Notifications
You must be signed in to change notification settings - Fork 4.2k
Refactor stub implementation of LTI Provider. BLD-601. #2029
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,237 @@ | ||
| """ | ||
| 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 = '' | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. a) add comment why we do this |
||
|
|
||
| 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 | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| """ | ||
| 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()) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -289,7 +289,7 @@ def get_outcome_service_url(self): | |
| 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' | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. please check sandbox with https |
||
| uri = '{scheme}://{host}{path}'.format( | ||
| scheme=scheme, | ||
| host=self.system.hostname, | ||
|
|
@@ -325,7 +325,11 @@ def get_lis_result_sourcedid(self): | |
| 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( | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this should be removed and new ticket should be created to fix that in future.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @auraz in case of remove doublequoting it will affect test system. Remember string formatting in logger in http.py. |
||
| 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): | ||
| """ | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This method could use more whitespace between the
ifelifandelsestatements.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
added readability