fix mocks, signatures and sig auth, add tests

Author: maxkahan

http_client/src/vonage_http_client/auth.py

@@ -1,8 +1,9 @@
-from base64 import b64encode
-from typing import Literal, Optional
 import hashlib
 import hmac
+from base64 import b64encode
+
 from time import time
+from typing import Literal, Optional
 
 from pydantic import validate_call
 from vonage_jwt.jwt import JwtClient
@@ -77,16 +78,16 @@ def create_basic_auth_string(self):
         )
         return f'Basic {hash}'
 
-    def sign_params(self, params: dict) -> dict:
-        """
-        Signs the provided message parameters using the signature secret provided to the `Auth` class.
-        If no signature secret is provided, the message parameters are signed using a simple MD5 hash.
+    def sign_params(self, params: dict) -> str:
+        """Signs the provided message parameters using the signature secret provided to the `Auth`
+        class. If no signature secret is provided, the message parameters are signed using a simple
+        MD5 hash.
 
         Args:
             params (dict): The message parameters to be signed.
 
         Returns:
-            dict: The signed message parameters.
+            str: A hexadecimal digest of the signed message parameters.
         """
 
         hasher = hmac.new(
@@ -96,6 +97,7 @@ def sign_params(self, params: dict) -> dict:
 
         if not params.get('timestamp'):
             params['timestamp'] = int(time())
+            print(params['timestamp'])
 
         for key in sorted(params):
             value = params[key]
@@ -105,14 +107,23 @@ def sign_params(self, params: dict) -> dict:
 
             hasher.update(f'&{key}={value}'.encode('utf-8'))
 
-        if self._signature_method is None:
-            hasher.update(self._signature_secret.encode())
         return hasher.hexdigest()
 
     @validate_call
     def check_signature(self, params: dict) -> bool:
+        """
+        Checks the signature hash of the given parameters.
+
+        Args:
+            params (dict): The parameters to check the signature for.
+                This should include the `sig` parameter which contains the
+                signature hash of the other parameters.
+
+        Returns:
+            bool: True if the signature is valid, False otherwise.
+        """
         signature = params.pop('sig', '').lower()
-        return hmac.compare_digest(signature, self._signature_secret(params))
+        return hmac.compare_digest(signature, self.sign_params(params))
 
     def _validate_input_combinations(
         self, api_key, api_secret, application_id, private_key, signature_secret

http_client/src/vonage_http_client/http_client.py

@@ -140,7 +140,7 @@ def make_request(
             with self._session.request(
                 request_type,
                 url,
-                params=params,
+                data=params,
                 headers=self._headers,
                 timeout=self._timeout,
             ) as response:

http_client/tests/test_auth.py

@@ -1,7 +1,6 @@
+import hashlib
 from os.path import dirname, join
 from unittest.mock import patch
-import hashlib
-import hmac
 
 from pydantic import ValidationError
 from pytest import raises
@@ -36,6 +35,8 @@ def test_create_auth_class_and_get_objects():
     assert auth.api_key == api_key
     assert auth.api_secret == api_secret
     assert type(auth._jwt_client) == JwtClient
+    assert auth._signature_secret == signature_secret
+    assert auth._signature_method == hashlib.sha256
 
 
 def test_create_new_auth_invalid_type():
@@ -63,6 +64,10 @@ def test_auth_init_with_invalid_combinations():
         Auth(api_secret=api_secret, application_id=application_id)
     with raises(InvalidAuthError):
         Auth(api_secret=api_secret, private_key=private_key)
+    with raises(InvalidAuthError):
+        Auth(application_id=application_id, signature_secret=signature_secret)
+    with raises(InvalidAuthError):
+        Auth(private_key=private_key, signature_secret=signature_secret)
 
 
 def test_auth_init_with_valid_api_key_and_api_secret():
@@ -110,96 +115,64 @@ def test_create_basic_auth_string():
     assert auth.create_basic_auth_string() == 'Basic cXdlcmFzZGY6MTIzNHF3ZXJhc2Rmenhjdg=='
 
 
-def test_auth_init_with_valid_combinations():
-    api_key = 'qwerasdf'
-    api_secret = '1234qwerasdfzxcv'
-    application_id = 'asdfzxcv'
-    private_key = 'dummy_private_key'
-    signature_secret = 'signature_secret'
-    signature_method = 'sha256'
-
+def test_sign_params():
     auth = Auth(
         api_key=api_key,
-        api_secret=api_secret,
-        application_id=application_id,
-        private_key=private_key,
         signature_secret=signature_secret,
         signature_method=signature_method,
     )
 
-    assert auth._api_key == api_key
-    assert auth._api_secret == api_secret
-    assert auth._jwt_client.application_id == application_id
-    assert auth._jwt_client.private_key == private_key
-    assert auth._signature_secret == signature_secret
-    assert auth._signature_method == hashlib.sha256
+    params = {'param1': 'value1', 'param2': 'value2', 'timestamp': 1234567890}
 
+    signed_params_hash = auth.sign_params(params)
 
-def test_auth_init_with_invalid_combinations():
-    api_key = 'qwerasdf'
-    api_secret = '1234qwerasdfzxcv'
-    application_id = 'asdfzxcv'
-    private_key = 'dummy_private_key'
-    signature_secret = 'signature_secret'
-    signature_method = 'invalid_method'
-
-    with patch('vonage_http_client.auth.hashlib') as mock_hashlib:
-        mock_hashlib.sha256.side_effect = AttributeError
-
-        auth = Auth(
-            api_key=api_key,
-            api_secret=api_secret,
-            application_id=application_id,
-            private_key=private_key,
-            signature_secret=signature_secret,
-            signature_method=signature_method,
-        )
-
-        assert auth._api_key == api_key
-        assert auth._api_secret == api_secret
-        assert auth._jwt_client is None
-        assert auth._signature_secret == signature_secret
-        assert auth._signature_method is None
+    assert (
+        signed_params_hash
+        == '280c4320703dbc98bfa22db676655ed2acfbfe8792b062ff7622e67f1183c287'
+    )
 
 
-def test_sign_params():
-    auth = Auth(signature_secret='signature_secret', signature_method='sha256')
+def test_sign_params_default_sig_method():
+    auth = Auth(api_key=api_key, signature_secret=signature_secret)
 
     params = {'param1': 'value1', 'param2': 'value2', 'timestamp': 1234567890}
 
-    signed_params = auth.sign_params(params)
+    signed_params_hash = auth.sign_params(params)
 
-    assert signed_params == 'asdf'
+    assert signed_params_hash == '724c2bf6ca423c36e20631b11d1c5753'
 
 
-def test_sign_params_default_sig_method():
-    auth = Auth()
+def test_sign_params_with_special_characters():
+    auth = Auth(api_key=api_key, signature_secret=signature_secret)
 
-    params = {'param1': 'value1', 'param2': 'value2', 'timestamp': 1234567890}
+    params = {'param1': 'value&1', 'param2': 'value=2', 'timestamp': 1234567890}
 
     signed_params = auth.sign_params(params)
 
-    assert signed_params == 'asdf'
+    assert signed_params == '2bbf0abafb2c55e5af6231513896a2ac'
 
 
-def test_sign_params_with_special_characters():
-    auth = Auth(signature_secret='signature_secret', signature_method='sha1')
+@patch('vonage_http_client.auth.time', return_value=12345)
+def test_sign_params_with_dynamic_timestamp(mock_time):
+    auth = Auth(api_key=api_key, signature_secret=signature_secret)
 
-    params = {'param1': 'value&1', 'param2': 'value=2', 'timestamp': 1234567890}
+    params = {'param1': 'value1', 'param2': 'value2'}
 
     signed_params = auth.sign_params(params)
 
-    assert signed_params == 'asdf'
+    assert signed_params == 'bc7e95bb4e341090b3a202a2885903a5'
 
 
-# def test_check_signature_with_valid_signature():
-#     auth = Auth(signature_secret='signature_secret')
-#     params = {'param1': 'value1', 'param2': 'value2', 'sig': 'valid_signature'}
-#     expected_signature = hmac.new(
-#         b'signature_secret', b'param1value1param2value2', hashlib.sha256
-#     ).hexdigest()
+def test_check_signature_with_valid_signature():
+    auth = Auth(api_key=api_key, signature_secret=signature_secret)
+    params = {
+        'param1': 'value1',
+        'param2': 'value2',
+        'sig': 'valid_signature',
+        'timestamp': 1234567890,
+    }
 
-#     assert auth.check_signature(params) == True
+    assert auth.check_signature(params) == True
 
 
 # def test_check_signature_with_invalid_signature():

http_client/tests/test_http_client.py

@@ -4,6 +4,7 @@
 import responses
 from pytest import raises
 from requests import Response
+from responses import matchers
 from vonage_http_client.auth import Auth
 from vonage_http_client.errors import (
     AuthenticationError,
@@ -97,19 +98,33 @@ def test_make_post_request():
 
 @responses.activate
 def test_make_post_request_with_signature():
+    params = {
+        'test': 'post request',
+        'testing': 'http client',
+        'timestamp': '1234567890',
+    }
+
     build_response(
-        path, 'POST', 'https://example.com/post_signed_params', 'example_post.json'
+        path,
+        'POST',
+        'https://example.com/post_signed_params',
+        'example_post.json',
+        match=[
+            matchers.urlencoded_params_matcher(
+                {
+                    **params,
+                    'api_key': 'asdfzxcv',
+                    'sig': '237b06fd1f994a9ec2f3283a4a0239f35b56d64639d6485b45cffedcb385b033',
+                }
+            )
+        ],
     )
     client = HttpClient(
         Auth(
             api_key='asdfzxcv', signature_secret='qwerasdfzxcv', signature_method='sha256'
         ),
         http_client_options={'api_host': 'example.com'},
     )
-    params = {
-        'test': 'post request',
-        'testing': 'http client',
-    }
 
     res = client.post(
         host='example.com',
@@ -118,8 +133,6 @@ def test_make_post_request_with_signature():
         auth_type='signature',
     )
     assert res['hello'] == 'world!'
-    print(responses.calls[0].request.url)
-    assert responses.calls[0].request.body == params
 
 
 @responses.activate

pants.toml

@@ -31,6 +31,7 @@ filter = [
   'vonage/src',
   'http_client/src',
   'number_insight_v2/src',
+  'sms/src',
   'utils/src',
   'testutils',
 ]

sms/src/vonage_sms/sms.py

@@ -5,7 +5,7 @@
 from pydantic import BaseModel, Field, field_validator, validate_call
 from vonage_http_client.http_client import HttpClient
 
-from .errors import SmsError, PartialFailureError
+from .errors import PartialFailureError, SmsError
 
 
 class SmsMessage(BaseModel):

testutils/testutils.py

@@ -10,6 +10,10 @@ def _load_mock_data(caller_file_path: str, mock_path: str):
         return file.read()
 
 
+def _filter_none_values(data: dict) -> dict:
+    return {k: v for (k, v) in data.items() if v is not None}
+
+
 @validate_call
 def build_response(
     file_path: str,
@@ -18,7 +22,18 @@ def build_response(
     mock_path: str = None,
     status_code: int = 200,
     content_type: str = 'application/json',
+    match: list = None,
 ):
-    print('file_path', file_path)
     body = _load_mock_data(file_path, mock_path) if mock_path else None
-    responses.add(method, url, body=body, status=status_code, content_type=content_type)
+    responses.add(
+        **_filter_none_values(
+            {
+                'method': method,
+                'url': url,
+                'body': body,
+                'status': status_code,
+                'content_type': content_type,
+                'match': match,
+            }
+        )
+    )