Merge pull request #8 from MastercardDevs/master

Add interceptor and test

Author: mickfGit

oauth1/oauth.py

@@ -28,6 +28,7 @@
 #
 from random import randint
 import time
+import json
 
 import oauth1.coreutils as util
 from OpenSSL import crypto
@@ -59,8 +60,11 @@ def get_oauth_parameters(self, uri, method, payload, consumer_key, signing_key):
         oauth_parameters.set_oauth_timestamp(OAuth.get_timestamp())
         oauth_parameters.set_oauth_signature_method("RSA-SHA256")
         oauth_parameters.set_oauth_version("1.0")
+
+        payload_str = json.dumps(payload) if type(payload) is dict else payload
+
         if method != "GET" and method != "DELETE" and method != "HEAD":
-            encoded_hash = util.base64_encode(util.sha256_encode(payload))
+            encoded_hash = util.base64_encode(util.sha256_encode(payload_str))
             oauth_parameters.set_oauth_body_hash(encoded_hash)
 
         # Get the base string

oauth1/signer_interceptor.py

@@ -0,0 +1,51 @@
+import json
+from functools import wraps
+from oauth1.oauth import OAuth
+from oauth1 import authenticationutils
+
+
+class SignerInterceptor(object):
+
+
+    def __init__(self, key_file, key_password, consumer_key):
+        """Load signing key."""
+        self.signing_key = authenticationutils.load_signing_key(key_file, key_password)
+        self.consumer_key = consumer_key
+
+
+    def oauth_signing(self, func):
+        """Decorator for API request. func is APIClient.request"""
+
+        @wraps(func)
+        def request_function(*args, **kwargs):  # pragma: no cover
+            in_body = kwargs.get("body", None)
+
+            auth_header = OAuth().get_authorization_header(args[1], args[0], in_body,
+                                                           self.consumer_key, self.signing_key)
+
+            in_headers = kwargs.get("headers", None)
+            if not in_headers:
+                in_headers = dict()
+                kwargs["headers"] = in_headers
+
+            in_headers["Authorization"] = auth_header
+
+            return func(*args, **kwargs)
+
+        return request_function
+      
+
+
+def add_signing_layer(self, api_client, key_file, key_password, consumer_key):
+    """Create and load configuration. Decorate APIClient.request with header signing"""
+
+    api_signer = SignerInterceptor(key_file, key_password, consumer_key)
+    api_client.request = api_signer.oauth_signing(api_client.request)
+
+def get_signing_layer(self, api_client):
+    return api_client.request
+
+
+
+ 
+ 

tests/test_interceptor.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-#
+#
+#
+# Copyright (c) 2019 MasterCard International Incorporated
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without modification, are
+# permitted provided that the following conditions are met:
+#
+# Redistributions of source code must retain the above copyright notice, this list of
+# conditions and the following disclaimer.
+# Redistributions in binary form must reproduce the above copyright notice, this list of
+# conditions and the following disclaimer in the documentation and/or other materials
+# provided with the distribution.
+# Neither the name of the MasterCard International Incorporated nor the names of its
+# contributors may be used to endorse or promote products derived from this software
+# without specific prior written permission.
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
+# SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
+# TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
+# OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER
+# IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING
+# IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+# SUCH DAMAGE.
+#
+import unittest
+import requests
+from urllib.parse import urlencode
+from oauth1.oauth import OAuth
+from oauth1.signer_interceptor import SignerInterceptor
+import oauth1.authenticationutils as authenticationutils
+from os.path import dirname, realpath, join, os
+from oauth1.signer import OAuthSigner
+from oauth1.signer_interceptor import add_signing_layer
+from oauth1.signer_interceptor import get_signing_layer
+
+
+class OAuthInterceptorTest(unittest.TestCase):
+
+
+    """ add an interceptor, check api client has changed """
+    def test_add_interceptor(self):
+        if os.path.exists('./test_key_container.p12'):
+            key_file = './test_key_container.p12'
+            key_password = "Password1"
+            consumer_key = 'uLXKmWNmIkzIGKfA2injnNQqpZaxaBSKxa3ixEVu2f283c95!33b9b2bd960147e387fa6f3f238f07170000000000000000'
+
+            signing_layer1 = get_signing_layer(self, requests)
+            
+            add_signing_layer(self, requests, key_file, key_password, consumer_key)
+
+            signing_layer2 = get_signing_layer(self, requests)
+
+            self.assertNotEqual(signing_layer1, signing_layer2)            
+
+
+        else:
+            print("Please add a ./test_key_container.p12 file to enable key tests")
+
+
+
+if __name__ == '__main__':
+    unittest.main()