Make signature configurable

Author: const-cloudinary

cloudinary/__init__.py

@@ -280,7 +280,7 @@ def __len__(self):
         return len(self.public_id) if self.public_id is not None else 0
 
     def validate(self):
-        return self.signature == self.get_expected_signature()
+        return utils.verify_api_response_signature(self.public_id, self.version, self.signature)
 
     def get_prep_value(self):
         if None in [self.public_id,
@@ -301,7 +301,7 @@ def get_presigned(self):
 
     def get_expected_signature(self):
         return utils.api_sign_request({"public_id": self.public_id, "version": self.version}, config().api_secret,
-                                      config().signature_algorithm)
+                                      config().signature_algorithm, signature_version=1)
 
     @property
     def url(self):

cloudinary/utils.py

@@ -619,9 +619,10 @@ def sign_request(params, options):
     if not api_secret:
         raise ValueError("Must supply api_secret")
     signature_algorithm = options.get("signature_algorithm", cloudinary.config().signature_algorithm)
+    signature_version = options.get("signature_version", cloudinary.config().signature_version)
 
     params = cleanup_params(params)
-    params["signature"] = api_sign_request(params, api_secret, signature_algorithm)
+    params["signature"] = api_sign_request(params, api_secret, signature_algorithm, signature_version)
     params["api_key"] = api_key
 
     return params
@@ -630,7 +631,7 @@ def sign_request(params, options):
 def api_sign_request(params_to_sign, api_secret, algorithm=SIGNATURE_SHA1, signature_version=2):
     """
     Signs API request parameters using the specified algorithm and signature version.
-    
+
     :param params_to_sign: Parameters to include in the signature
     :param api_secret: API secret key
     :param algorithm: Signature algorithm (default: SHA1)
@@ -646,7 +647,7 @@ def api_sign_request(params_to_sign, api_secret, algorithm=SIGNATURE_SHA1, signa
 def api_string_to_sign(params_to_sign, signature_version=2):
     """
     Generates a string to be signed for API requests.
-    
+
     :param params_to_sign: Parameters to include in the signature
     :param signature_version: Version of signature algorithm to use:
         - Version 1: Original behavior without parameter encoding
@@ -662,19 +663,19 @@ def api_string_to_sign(params_to_sign, signature_version=2):
                 value = str(v).lower()
             else:
                 value = str(v)
-            
+
             param_string = k + "=" + value
             if signature_version >= 2:
                 param_string = _encode_param(param_string)
             params.append(param_string)
-    
+
     return "&".join(sorted(params))
 
 
 def _encode_param(value):
     """
     Encodes a parameter for safe inclusion in URL query strings.
-    
+
     Specifically replaces "&" characters with their percent-encoded equivalent "%26"
     to prevent them from being interpreted as parameter separators in URL query strings.
 
@@ -1613,7 +1614,7 @@ def verify_api_response_signature(public_id, version, signature, algorithm=None)
     :param version:   The version of the asset as returned in the API response
     :param signature: Actual signature. Can be retrieved from the X-Cld-Signature header
     :param algorithm: Name of hashing algorithm to use for calculation of HMACs.
-                      By default uses `cloudinary.config().signature_algorithm`
+                      By default, uses `cloudinary.config().signature_algorithm`
 
     :return: Boolean result of the validation
     """
@@ -1641,7 +1642,7 @@ def verify_notification_signature(body, timestamp, signature, valid_for=7200, al
     :param signature: Actual signature. Can be retrieved from the X-Cld-Signature header
     :param valid_for: The desired time in seconds for considering the request valid
     :param algorithm: Name of hashing algorithm to use for calculation of HMACs.
-                      By default uses `cloudinary.config().signature_algorithm`
+                      By default, uses `cloudinary.config().signature_algorithm`
 
     :return: Boolean result of the validation
     """

test/test_utils.py

@@ -78,7 +78,8 @@ def setUp(self):
                           cname=None,  # for these tests without actual upload, we ignore cname
                           api_key="a", api_secret="b",
                           secure_distribution=None,
-                          private_cdn=False)
+                          private_cdn=False,
+                          signature_version=2)
 
     def __test_cloudinary_url(self, public_id=TEST_ID, options=None, expected_url=None, expected_options=None):
         if expected_options is None:
@@ -1460,31 +1461,31 @@ def test_api_sign_request_prevents_parameter_smuggling(self):
         """Should prevent parameter smuggling via & characters in parameter values"""
         # Test with notification_url containing & characters
         params_with_ampersand = {
-            "cloud_name": API_SIGN_REQUEST_CLOUD_NAME, 
+            "cloud_name": API_SIGN_REQUEST_CLOUD_NAME,
             "timestamp": 1568810420,
             "notification_url": "https://fake.com/callback?a=1&tags=hello,world"
         }
-        
+
         signature_with_ampersand = api_sign_request(params_with_ampersand, API_SIGN_REQUEST_TEST_SECRET)
-        
+
         # Test that attempting to smuggle parameters by splitting the notification_url fails
         params_smuggled = {
             "cloud_name": API_SIGN_REQUEST_CLOUD_NAME,
-            "timestamp": 1568810420, 
+            "timestamp": 1568810420,
             "notification_url": "https://fake.com/callback?a=1",
             "tags": "hello,world"  # This would be smuggled if & encoding didn't work
         }
-        
+
         signature_smuggled = api_sign_request(params_smuggled, API_SIGN_REQUEST_TEST_SECRET)
-        
+
         # The signatures should be different, proving that parameter smuggling is prevented
         self.assertNotEqual(signature_with_ampersand, signature_smuggled,
                            "Signatures should be different to prevent parameter smuggling")
-        
+
         # Verify the expected signature for the properly encoded case
         expected_signature = "4fdf465dd89451cc1ed8ec5b3e314e8a51695704"
         self.assertEqual(expected_signature, signature_with_ampersand)
-        
+
         # Verify the expected signature for the smuggled parameters case
         expected_smuggled_signature = "7b4e3a539ff1fa6e6700c41b3a2ee77586a025f9"
         self.assertEqual(expected_smuggled_signature, signature_smuggled)
@@ -1493,23 +1494,23 @@ def test_api_sign_request_signature_versions(self):
         """Should use signature version 1 (without parameter encoding) for backward compatibility"""
         public_id_with_ampersand = 'tests/logo&version=2'
         test_version = 1
-        
+
         expected_signature_v1 = api_sign_request(
             {'public_id': public_id_with_ampersand, 'version': test_version},
             API_SIGN_REQUEST_TEST_SECRET,
             cloudinary.utils.SIGNATURE_SHA1,
             signature_version=1
         )
-        
+
         expected_signature_v2 = api_sign_request(
             {'public_id': public_id_with_ampersand, 'version': test_version},
             API_SIGN_REQUEST_TEST_SECRET,
             cloudinary.utils.SIGNATURE_SHA1,
             signature_version=2
         )
-        
+
         self.assertNotEqual(expected_signature_v1, expected_signature_v2)
-        
+
         # verify_api_response_signature should use version 1 for backward compatibility
         with patch('cloudinary.config', return_value=cloudinary.config(api_secret=API_SIGN_REQUEST_TEST_SECRET)):
             self.assertTrue(
@@ -1519,7 +1520,7 @@ def test_api_sign_request_signature_versions(self):
                     expected_signature_v1
                 )
             )
-            
+
             self.assertFalse(
                 verify_api_response_signature(
                     public_id_with_ampersand,
@@ -1528,6 +1529,68 @@ def test_api_sign_request_signature_versions(self):
                 )
             )
 
+    def test_signature_version_config_support(self):
+        """Should use signature_version from config and produce different signatures for v1 vs v2"""
+        # Use params with & characters to show the encoding difference between versions
+        params = {'public_id': 'test&image', 'notification_url': 'https://example.com/callback?param=value&other=data'}
+
+        # Test with config signature_version = 1
+        cloudinary.config().signature_version = 1
+
+        # Test sign_request function uses config values
+        options_with_config = {'api_key': 'test_key', 'api_secret': API_SIGN_REQUEST_TEST_SECRET}
+        signed_params_config_v1 = cloudinary.utils.sign_request(params.copy(), options_with_config)
+
+        # Test explicit signature version
+        options_explicit_v1 = options_with_config.copy()
+        options_explicit_v1['signature_version'] = 1
+        signed_params_explicit_v1 = cloudinary.utils.sign_request(params.copy(), options_explicit_v1)
+
+        self.assertEqual(signed_params_config_v1['signature'], signed_params_explicit_v1['signature'])
+
+        # Test with config signature_version = 2
+        cloudinary.config().signature_version = 2
+
+        signed_params_config_v2 = cloudinary.utils.sign_request(params.copy(), options_with_config)
+
+        options_explicit_v2 = options_with_config.copy()
+        options_explicit_v2['signature_version'] = 2
+        signed_params_explicit_v2 = cloudinary.utils.sign_request(params.copy(), options_explicit_v2)
+
+        self.assertEqual(signed_params_config_v2['signature'], signed_params_explicit_v2['signature'])
+
+        # Verify that v1 and v2 actually produce different signatures due to parameter encoding
+        self.assertNotEqual(signed_params_config_v1['signature'], signed_params_config_v2['signature'],
+                          "Signature v1 and v2 should be different for parameters with & characters")
+
+    def test_sign_request_with_signature_version(self):
+        """Should support signature_version parameter in sign_request function"""
+        params = {'public_id': 'test_image', 'version': 1234}
+        options = {'api_key': 'test_key', 'api_secret': API_SIGN_REQUEST_TEST_SECRET}
+
+        # Test with signature_version in options
+        options_v1 = options.copy()
+        options_v1['signature_version'] = 1
+        signed_params_v1 = cloudinary.utils.sign_request(params.copy(), options_v1)
+
+        options_v2 = options.copy()
+        options_v2['signature_version'] = 2
+        signed_params_v2 = cloudinary.utils.sign_request(params.copy(), options_v2)
+
+        # The signatures should be different for different versions (for params with & characters)
+        # For these simple params without & they might be the same, but let's test the structure
+        self.assertIn('signature', signed_params_v1)
+        self.assertIn('signature', signed_params_v2)
+        self.assertIn('api_key', signed_params_v1)
+        self.assertIn('api_key', signed_params_v2)
+
+        # Test that signature_version is passed through correctly
+        expected_sig_v1 = api_sign_request(params, API_SIGN_REQUEST_TEST_SECRET, cloudinary.utils.SIGNATURE_SHA1, 1)
+        expected_sig_v2 = api_sign_request(params, API_SIGN_REQUEST_TEST_SECRET, cloudinary.utils.SIGNATURE_SHA1, 2)
+
+        self.assertEqual(signed_params_v1['signature'], expected_sig_v1)
+        self.assertEqual(signed_params_v2['signature'], expected_sig_v2)
+
 
 if __name__ == '__main__':
     unittest.main()