simplifying code, thanks Carlson.

Add client certificate support Moved add cert to post_message methods.
mitodl · Dec 8, 2014 · 1659482 · 1659482
1 parent 4c1ec19
commit 1659482
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 5 deletions.
diff --git a/pylti/__init__.py b/pylti/__init__.py
@@ -3,4 +3,4 @@
 PyLTI is module that implements IMS LTI in python
 The API uses decorators to wrap function with LTI functionality.
 """
-VERSION = "0.1.3"  # pragma: no cover
+VERSION = "0.1.4"  # pragma: no cover
diff --git a/pylti/common.py b/pylti/common.py
@@ -82,6 +82,23 @@ def lookup_consumer(self, key):
             return None
         return oauth.OAuthConsumer(key, secret)
 
+    def lookup_cert(self, key):
+        """
+        Search through keys
+        """
+        if not self.consumers:
+            log.critical(("No consumers defined in settings."
+                          "Have you created a configuration file?"))
+            return None
+
+        consumer = self.consumers.get(key)
+        if not consumer:
+            log.info("Did not find consumer, using key: %s ", key)
+            return None
+
+        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
@@ -182,16 +199,22 @@ def post_message(consumers, lti_key, url, body):
     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)
+
     secret = lti_consumer.secret
 
     consumer = oauth2.Consumer(key=lti_key, secret=secret)
     client = oauth2.Client(consumer)
+
+    if lti_cert:
+        client.add_certificate(key=lti_cert, cert=lti_cert, domain='')
+
     (response, content) = _post_patched_request(
         body,
         client,
         url,
         method,
-        content_type
+        content_type,
     )
 
     log.debug("key {}".format(lti_key))
@@ -220,16 +243,22 @@ def post_message2(consumers, lti_key, url, body,
     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)
+
     secret = lti_consumer.secret
 
     consumer = oauth2.Consumer(key=lti_key, secret=secret)
     client = oauth2.Client(consumer)
+
+    if lti_cert:
+        client.add_certificate(key=lti_cert, cert=lti_cert, domain='')
+
     (response, content) = _post_patched_request(
         body,
         client,
         url,
         method,
-        content_type
+        content_type,
     )
 
     log.debug("POST MESSAGE 2")

diff --git a/pylti/tests/test_common.py b/pylti/tests/test_common.py
@@ -13,6 +13,7 @@
     verify_request_common,
     LTIException,
     post_message,
+    post_message2,
     generate_request_xml
 )
 
@@ -63,14 +64,18 @@ def test_ltioauthdatastore(self):
             "key1": {"secret": "secret1"},
             "key2": {"secret": "secret2"},
             "key3": {"secret": "secret3"},
-            "keyNS": {"test": "test"}
+            "keyNS": {"test": "test"},
+            "keyWCert": {"secret": "secret", "cert": "cert", "key": "key"}
         }
         store = LTIOAuthDataStore(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")
+        self.assertEqual(store.lookup_cert("keyWCert"), "cert")
         self.assertIsNone(store.lookup_consumer("key4"))
+        self.assertIsNone(store.lookup_cert("key4"))
         self.assertIsNone(store.lookup_consumer("keyNS"))
+        self.assertIsNone(store.lookup_cert("keyNS"))
 
     def test_ltioauthdatastore_no_consumers(self):
         """
@@ -79,6 +84,7 @@ def test_ltioauthdatastore_no_consumers(self):
         """
         store = LTIOAuthDataStore(None)
         self.assertIsNone(store.lookup_consumer("key1"))
+        self.assertIsNone(store.lookup_cert("key1"))
 
     def test_verify_request_common(self):
         """
@@ -170,13 +176,17 @@ def request_callback(request, cburi, headers):
 
         httpretty.register_uri(httpretty.POST, uri, body=request_callback)
         consumers = {
-            "__consumer_key__": {"secret": "__lti_secret__"}
+            "__consumer_key__": {"secret": "__lti_secret__",
+                                 "cert": "cert", "key": "key"}
         }
         body = generate_request_xml('message_identifier_id', 'operation',
                                     'lis_result_sourcedid', '1.0')
         ret = post_message(consumers, "__consumer_key__", uri, body)
         self.assertTrue(ret)
 
+        ret = post_message2(consumers, "__consumer_key__", uri, body)
+        self.assertTrue(ret)
+
     def test_generate_xml(self):
         """
         Generated post XML is valid