Full Python 3 compatibility

.travis.yml

@@ -5,6 +5,7 @@ python:
  - 2.7
 install:
  - python setup.py install --quiet
+ - pip install --upgrade -r test_requirements.txt
  - pip install coveralls
 script:
  - python setup.py test --coverage --pep8 --flakes

README.rst

@@ -38,21 +38,21 @@ Flask `mit_lti_flask_sample
 
 Dependencies:
 =============
-* Python 2.7+
-* oauth 1.0.1+
-* oauth2 1.5.211+
+* Python 2.7+ or Python 3.4+
+* oauth2 1.9.0+
 * httplib2 0.9+
+* six 1.10.0+
 
 Development dependencies:
 =========================
 * Flask 0.10.1
 * httpretty 0.8.3
 * oauthlib 0.6.3
-* pyflakes 0.8.1
-* pytest 2.6.3
+* pyflakes 1.2.3
+* pytest 2.9.2
 * pytest-cache 1.0
-* pytest-cov 1.8.0
-* pytest-flakes 0.2
+* pytest-cov 2.3.0
+* pytest-flakes 1.0.1
 * pytest-pep8 1.0.6
 * sphinx 1.2.3
 

pylti/common.py

@@ -4,12 +4,15 @@
 """
 
 from __future__ import absolute_import
+
 import logging
 
 import oauth2
-import oauth.oauth as oauth
 from xml.etree import ElementTree as etree
 
+from oauth2 import STRING_TYPES
+from six.moves.urllib.parse import urlparse, urlencode
+
 log = logging.getLogger('pylti.common') # pylint: disable=invalid-name
 
 LTI_PROPERTY_LIST = [
@@ -49,18 +52,17 @@
 LTI_REQUEST_TYPE = [u'any', u'initial', u'session']
 
 
-class LTIOAuthDataStore(oauth.OAuthDataStore):
- # pylint: disable=abstract-method
+class LTIOAuthServer(oauth2.Server):
  """
  Largely taken from reference implementation
  for app engine at https://code.google.com/p/ims-dev/
  """
 
- def __init__(self, consumers):
+ def __init__(self, consumers, signature_methods=None):
  """
- Create OAuth store
+ Create OAuth server
  """
- oauth.OAuthDataStore.__init__(self)
+ super(LTIOAuthServer, self).__init__(signature_methods)
  self.consumers = consumers
 
  def lookup_consumer(self, key):
@@ -82,7 +84,7 @@ def lookup_consumer(self, key):
  log.critical(('Consumer %s, is missing secret'
  'in settings file, and needs correction.'), key)
  return None
- return oauth.OAuthConsumer(key, secret)
+ return oauth2.Consumer(key, secret)
 
  def lookup_cert(self, key):
  """
@@ -101,15 +103,6 @@ def lookup_cert(self, key):
  cert = consumer.get('cert', None)
  return cert
 
- def lookup_nonce(self, oauth_consumer, oauth_token, nonce):
- """
- Lookup nonce should check if nonce was already used
- by this consumer in the past.
- Reusing nonce is bad: http://cwe.mitre.org/data/definitions/323.html
- Not implemented.
- """
- return None
-
 
 class LTIException(Exception):
  """
@@ -155,11 +148,10 @@ def _post_patched_request(consumers, lti_key, body,
  :return: response
  """
  # pylint: disable=too-many-locals, too-many-arguments
- oauth_store = LTIOAuthDataStore(consumers)
- oauth_server = oauth.OAuthServer(oauth_store)
- oauth_server.add_signature_method(oauth.OAuthSignatureMethod_HMAC_SHA1())
- lti_consumer = oauth_store.lookup_consumer(lti_key)
- lti_cert = oauth_store.lookup_cert(lti_key)
+ oauth_server = LTIOAuthServer(consumers)
+ oauth_server.add_signature_method(SignatureMethod_HMAC_SHA1_Unicode())
+ lti_consumer = oauth_server.lookup_consumer(lti_key)
+ lti_cert = oauth_server.lookup_cert(lti_key)
 
  secret = lti_consumer.secret
 
@@ -190,7 +182,7 @@ def my_normalize(self, headers):
  response, content = client.request(
  url,
  method,
- body=body,
+ body=body.encode('utf-8'),
  headers={'Content-Type': content_type})
 
  http = httplib2.Http
@@ -227,7 +219,7 @@ def post_message(consumers, lti_key, url, body):
  content_type,
  )
 
- is_success = "<imsx_codeMajor>success</imsx_codeMajor>" in content
+ is_success = b"<imsx_codeMajor>success</imsx_codeMajor>" in content
  log.debug("is success %s", is_success)
  return is_success
 
@@ -277,18 +269,17 @@ def verify_request_common(consumers, url, method, headers, params):
  log.debug("headers %s", headers)
  log.debug("params %s", params)
 
- oauth_store = LTIOAuthDataStore(consumers)
- oauth_server = oauth.OAuthServer(oauth_store)
+ oauth_server = LTIOAuthServer(consumers)
  oauth_server.add_signature_method(
- oauth.OAuthSignatureMethod_PLAINTEXT())
+ SignatureMethod_PLAINTEXT_Unicode())
  oauth_server.add_signature_method(
- oauth.OAuthSignatureMethod_HMAC_SHA1())
+ SignatureMethod_HMAC_SHA1_Unicode())
 
  # Check header for SSL before selecting the url
  if headers.get('X-Forwarded-Proto', 'http') == 'https':
  url = url.replace('http:', 'https:', 1)
 
- oauth_request = oauth.OAuthRequest.from_request(
+ oauth_request = Request_Fix_Duplicate.from_request(
  method,
  url,
  headers=dict(headers),
@@ -301,9 +292,12 @@ def verify_request_common(consumers, url, method, headers, params):
  'or request')
  try:
  # pylint: disable=protected-access
- consumer = oauth_server._get_consumer(oauth_request)
- oauth_server._check_signature(oauth_request, consumer, None)
- except oauth.OAuthError:
+ oauth_consumer_key = oauth_request.get_parameter('oauth_consumer_key')
+ consumer = oauth_server.lookup_consumer(oauth_consumer_key)
+ if not consumer:
+ raise oauth2.Error('Invalid consumer.')
+ oauth_server.verify_request(oauth_request, consumer, None)
+ except oauth2.Error:
  # Rethrow our own for nice error handling (don't print
  # error message as it will contain the key
  raise LTIException("OAuth error: Please check your key and secret")
@@ -350,7 +344,110 @@ def generate_request_xml(message_identifier_id, operation,
  text_string = etree.SubElement(result_score, 'textString')
  text_string.text = score.__str__()
  ret = "<?xml version='1.0' encoding='utf-8'?>\n{}".format(
- etree.tostring(root, encoding='utf-8'))
+ etree.tostring(root, encoding='utf-8').decode('utf-8'))
 
  log.debug("XML Response: \n%s", ret)
  return ret
+
+
+class SignatureMethod_HMAC_SHA1_Unicode(oauth2.SignatureMethod_HMAC_SHA1):
+ """
+ Temporary workaround for
+ https:/joestump/python-oauth2/issues/207
+
+ Original code is Copyright (c) 2007 Leah Culver, MIT license.
+ """
+
+ def check(self, request, consumer, token, signature):
+ """
+ Returns whether the given signature is the correct signature for
+ the given consumer and token signing the given request.
+ """
+ built = self.sign(request, consumer, token)
+ if isinstance(signature, STRING_TYPES):
+ signature = signature.encode("utf8")
+ return built == signature
+
+
+class SignatureMethod_PLAINTEXT_Unicode(oauth2.SignatureMethod_PLAINTEXT):
+ """
+ Temporary workaround for
+ https:/joestump/python-oauth2/issues/207
+
+ Original code is Copyright (c) 2007 Leah Culver, MIT license.
+ """
+
+ def check(self, request, consumer, token, signature):
+ """
+ Returns whether the given signature is the correct signature for
+ the given consumer and token signing the given request.
+ """
+ built = self.sign(request, consumer, token)
+ if isinstance(signature, STRING_TYPES):
+ signature = signature.encode("utf8")
+ return built == signature
+
+
+class Request_Fix_Duplicate(oauth2.Request):
+ """
+ Temporary workaround for
+ https:/joestump/python-oauth2/pull/197
+
+ Original code is Copyright (c) 2007 Leah Culver, MIT license.
+ """
+
+ def get_normalized_parameters(self):
+ """
+ Return a string that contains the parameters that must be signed.
+ """
+ items = []
+ for key, value in self.items():
+ if key == 'oauth_signature':
+ continue
+ # 1.0a/9.1.1 states that kvp must be sorted by key, then by value,
+ # so we unpack sequence values into multiple items for sorting.
+ if isinstance(value, STRING_TYPES):
+ items.append(
+ (oauth2.to_utf8_if_string(key), oauth2.to_utf8(value))
+ )
+ else:
+ try:
+ value = list(value)
+ except TypeError as e:
+ assert 'is not iterable' in str(e)
+ items.append(
+ (oauth2.to_utf8_if_string(key),
+ oauth2.to_utf8_if_string(value))
+ )
+ else:
+ items.extend(
+ (oauth2.to_utf8_if_string(key),
+ oauth2.to_utf8_if_string(item))
+ for item in value
+ )
+
+ # Include any query string parameters from the provided URL
+ query = urlparse(self.url)[4]
+ url_items = self._split_url_string(query).items()
+ url_items = [
+ (oauth2.to_utf8(k), oauth2.to_utf8_optional_iterator(v))
+ for k, v in url_items if k != 'oauth_signature'
+ ]
+
+ # Merge together URL and POST parameters.
+ # Eliminates parameters duplicated between URL and POST.
+ items_dict = {}
+ for k, v in items:
+ items_dict.setdefault(k, []).append(v)
+ for k, v in url_items:
+ if not (k in items_dict and v in items_dict[k]):
+ items.append((k, v))
+
+ items.sort()
+
+ encoded_str = urlencode(items, True)
+ # Encode signature parameters per Oauth Core 1.0 protocol
+ # spec draft 7, section 3.6
+ # (http://tools.ietf.org/html/draft-hammer-oauth-07#section-3.6)
+ # Spaces must be encoded with "%20" instead of "+"
+ return encoded_str.replace('+', '%20').replace('%7E', '~')

pylti/flask.py

@@ -214,9 +214,9 @@ def response_url(self):
  urls = app_config.get('PYLTI_URL_FIX', dict())
  # url remapping is useful for using devstack
  # devstack reports httpS://localhost:8000/ and listens on HTTP
- for prefix, mapping in urls.iteritems():
+ for prefix, mapping in urls.items():
  if url.startswith(prefix):
- for _from, _to in mapping.iteritems():
+ for _from, _to in mapping.items():
  url = url.replace(_from, _to)
  return url
 

pylti/tests/test_common.py

@@ -7,12 +7,12 @@
 
 import httpretty
 import oauthlib.oauth1
-from urlparse import urlparse, parse_qs
-import urllib
+
+from six.moves.urllib.parse import urlencode, urlparse, parse_qs
 
 import pylti
 from pylti.common import (
- LTIOAuthDataStore,
+ LTIOAuthServer,
  verify_request_common,
  LTIException,
  post_message,
@@ -58,9 +58,9 @@ def test_version():
  """
  semantic_version.Version(pylti.VERSION)
 
- def test_lti_oauth_data_store(self):
+ def test_lti_oauth_server(self):
  """
- Tests that LTIOAuthDataStore works
+ Tests that LTIOAuthServer works
  """
  consumers = {
  "key1": {"secret": "secret1"},
@@ -69,7 +69,7 @@ def test_lti_oauth_data_store(self):
  "keyNS": {"test": "test"},
  "keyWCert": {"secret": "secret", "cert": "cert"},
  }
- store = LTIOAuthDataStore(consumers)
+ store = LTIOAuthServer(consumers)
  self.assertEqual(store.lookup_consumer("key1").secret, "secret1")
  self.assertEqual(store.lookup_consumer("key2").secret, "secret2")
  self.assertEqual(store.lookup_consumer("key3").secret, "secret3")
@@ -79,12 +79,12 @@ def test_lti_oauth_data_store(self):
  self.assertIsNone(store.lookup_consumer("keyNS"))
  self.assertIsNone(store.lookup_cert("keyNS"))
 
- def test_lti_oauth_data_store_no_consumers(self):
+ def test_lti_oauth_server_no_consumers(self):
  """
  If consumers are not given it there are no consumer to return.
  """
 
- store = LTIOAuthDataStore(None)
+ store = LTIOAuthServer(None)
  self.assertIsNone(store.lookup_consumer("key1"))
  self.assertIsNone(store.lookup_cert("key1"))
 
@@ -248,7 +248,7 @@ def generate_oauth_request(url_to_sign=None):
  u'6ac8/handler_noauth'
  u'/grade_handler',
  'lti_message_type': u'basic-lti-launch-request'}
- urlparams = urllib.urlencode(params)
+ urlparams = urlencode(params)
 
  client = oauthlib.oauth1.Client('__consumer_key__',
  client_secret='__lti_secret__',
@@ -261,6 +261,6 @@ def generate_oauth_request(url_to_sign=None):
  url_parts = urlparse(signature[0])
  query_string = parse_qs(url_parts.query, keep_blank_values=True)
  verify_params = dict()
- for key, value in query_string.iteritems():
+ for key, value in query_string.items():
  verify_params[key] = value[0]
  return consumers, method, url, verify_params, params