added warning on missing oauth layer in encryption decorator

warning tests

Author: pqlMC

client_encryption/api_encryption.py

@@ -1,5 +1,6 @@
 import json
 from functools import wraps
+from warnings import warn
 from client_encryption.field_level_encryption_config import FieldLevelEncryptionConfig
 from client_encryption.session_key_params import SessionKeyParams
 from client_encryption.field_level_encryption import encrypt_payload, decrypt_payload
@@ -34,6 +35,7 @@ def request_function(*args, **kwargs):
 
             return response
 
+        request_function.__fle__ = True
         return request_function
 
     def _encrypt_payload(self, headers, body):
@@ -88,3 +90,20 @@ def add_encryption_layer(api_client, encryption_conf_file):
 
     api_encryption = ApiEncryption(encryption_conf_file)
     api_client.request = api_encryption.field_encryption(api_client.request)
+
+    __check_oauth(api_client)  # warn the user if authentication layer is missing/not set
+
+
+def __check_oauth(api_client):
+    try:
+        oauth_layer = getattr(api_client.request, "__wrapped__").__oauth__
+        if not oauth_layer or type(oauth_layer) is not bool:
+            __oauth_warn()
+    except AttributeError:
+        __oauth_warn()
+
+
+def __oauth_warn():
+    warn("No signing layer detected. Request will be only encrypted without being signed. "
+         "Please refer to "
+         "https://github.com/Mastercard/client-encryption-python#integrating-with-mastercard-oauth1-signer-module")

tests/test_api_encryption.py

@@ -1,5 +1,5 @@
 import unittest
-from unittest.mock import patch
+from unittest.mock import patch, Mock
 import json
 from tests.utils.api_encryption_test_utils import MockApiClient, MockService
 from tests import get_config_for_test, TEST_CONFIG
@@ -328,3 +328,35 @@ def test_add_header_encryption_layer_get(self):
         self.assertIn("secret", response.data["data"])
         self.assertEqual([53, 84, 75], response.data["data"]["secret"])
         self.assertDictEqual({"Content-Type": "application/json"}, response.getheaders())
+
+    @patch('client_encryption.api_encryption.__oauth_warn')
+    def test_add_encryption_layer_oauth_set(self, __oauth_warn):
+        test_client = MockApiClient()
+        to_test.add_encryption_layer(test_client, self._json_config)
+
+        assert not __oauth_warn.called
+
+    def test_add_encryption_layer_missing_oauth_layer_warning(self):
+        test_client = Mock()
+
+        # no __oauth__ flag
+        with self.assertWarns(UserWarning):
+            to_test.add_encryption_layer(test_client, self._json_config)
+
+    def test_add_encryption_layer_wrong_oauth_layer_flag_warning(self):
+        test_client = Mock()
+
+        # __oauth__ is None
+        test_client.request.__oauth__ = None
+        with self.assertWarns(UserWarning):
+            to_test.add_encryption_layer(test_client, self._json_config)
+
+        # __oauth__ is False
+        test_client.request.__oauth__ = False
+        with self.assertWarns(UserWarning):
+            to_test.add_encryption_layer(test_client, self._json_config)
+
+        # __oauth__ is not a boolean
+        test_client.request.__oauth__ = 5
+        with self.assertWarns(UserWarning):
+            to_test.add_encryption_layer(test_client, self._json_config)

tests/utils/api_encryption_test_utils.py

@@ -1,11 +1,22 @@
 from unittest.mock import Mock
+from functools import wraps
 import json
 from tests import get_config_for_test
 import client_encryption.field_level_encryption as encryption
 import client_encryption.field_level_encryption_config as encryption_config
 from client_encryption.session_key_params import SessionKeyParams
 
 
+def mock_signing(func):
+    """Decorator to mock signing layer and avoid warnings."""
+    @wraps(func)
+    def request_function(*args, **kwargs):
+        return func(*args, **kwargs)
+
+    request_function.__oauth__ = True
+    return request_function
+
+
 class MockService(object):
 
     def __init__(self, api_client=None):
@@ -41,6 +52,7 @@ def __init__(self, configuration=None, header_name=None, header_value=None,
         json_config["paths"]["$"]["toDecrypt"] = {"encryptedData": "data"}
         self._config = encryption_config.FieldLevelEncryptionConfig(json_config)
 
+    @mock_signing
     def request(self, method, url, query_params=None, headers=None,
                 post_params=None, body=None, _preload_content=True,
                 _request_timeout=None):