From 161ff09f786ea5fa2b302e4c23b5b860e5a53f32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Furkan=20=C3=87oban?= Date: Fri, 2 Oct 2026 20:06:02 +0300 Subject: [PATCH] fix: return verified boolean in verify_signature, prevent timing attacks, and ensure thread-safe headers - Fix verify_signature to return verified boolean rather than None, enabling callers to properly check payment notification validity. - Use hmac.compare_digest for constant-time comparison in verify_signature to guard against timing attacks. - Make calculate_hmac_sha256_signature resilient against TypeErrors when params contain numbers or None. - Prevent multi-threading race conditions in concurrent web frameworks (FastAPI/Django) by returning per-request copies rather than mutating class-level header dict in-place. - Sanitize base_url host in connect() to handle both scheme-less and https:// prefixes. - Add comprehensive unit test suite in tests/ covering signature verification, thread-safety, and helpers. - Update GitHub Actions workflow to run automated unit tests across Python 3.9-3.13 matrix. --- .github/workflows/github_pull_request.yml | 5 +- iyzipay/iyzipay_resource.py | 22 +++-- tests/__init__.py | 1 + tests/test_iyzipay_resource.py | 107 ++++++++++++++++++++++ tests/test_pki_builder.py | 45 +++++++++ 5 files changed, 170 insertions(+), 10 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/test_iyzipay_resource.py create mode 100644 tests/test_pki_builder.py diff --git a/.github/workflows/github_pull_request.yml b/.github/workflows/github_pull_request.yml index 45fea81..5b58abb 100644 --- a/.github/workflows/github_pull_request.yml +++ b/.github/workflows/github_pull_request.yml @@ -25,4 +25,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install unittest2 + pip install . + - name: Run unit tests + run: | + python -m unittest discover -s tests -v diff --git a/iyzipay/iyzipay_resource.py b/iyzipay/iyzipay_resource.py index 0fcd46d..175879f 100644 --- a/iyzipay/iyzipay_resource.py +++ b/iyzipay/iyzipay_resource.py @@ -23,26 +23,30 @@ def __init__(self): def verify_signature(self, params, secret_key, signature): calculated_signature = self.calculate_hmac_sha256_signature(params, secret_key) - verified = signature == calculated_signature + verified = hmac.compare_digest(str(signature), str(calculated_signature)) print('Signature verified:', verified) + return verified def connect(self, method, url, options, request_body_dict=None): - connection = self.httplib.HTTPSConnection(options['base_url']) - body_str = json.dumps(request_body_dict) - header = self.get_http_header(url, options, body_str) + base_url = options.get('base_url', '') if options else '' + host = base_url.replace('https://', '').replace('http://', '').split('/')[0] + connection = self.httplib.HTTPSConnection(host) + body_str = json.dumps(request_body_dict) if request_body_dict is not None else None + header = self.get_http_header(url, options, body_str if body_str is not None else '') connection.request(method, url, body_str, header) return connection.getresponse() def get_http_header(self, url, options=None, body_str=None): random_str = self.generate_random_string(self.RANDOM_STRING_SIZE) - self.header.update({'x-iyzi-rnd': random_str}) return self.get_http_header_v2(url, options, random_str, body_str) def get_http_header_v2(self, url, options, random_str, body_str): url = url.split('?')[0] hashed_v2_str = self.generate_v2_hash(options['api_key'], url, options['secret_key'], random_str, body_str) - self.header.update({'Authorization': 'IYZWSv2 %s' % hashed_v2_str}) - return self.header + header = self.header.copy() + header['x-iyzi-rnd'] = random_str + header['Authorization'] = 'IYZWSv2 %s' % hashed_v2_str + return header @staticmethod def strip_zero(number): @@ -52,7 +56,7 @@ def strip_zero(number): @staticmethod def calculate_hmac_sha256_signature(params, secret_key): secret_key = bytes(secret_key.encode('utf-8')) - msg = ':'.join(params).encode('utf-8') + msg = ':'.join("" if p is None else str(p) for p in params).encode('utf-8') hmac_obj = hmac.new(secret_key, digestmod=hashlib.sha256) hmac_obj.update(msg) @@ -61,7 +65,7 @@ def calculate_hmac_sha256_signature(params, secret_key): @staticmethod def generate_v2_hash(api_key, url, secret_key, random_str, body_str): secret_key = bytes(secret_key.encode('utf-8')) - msg = (random_str + url + body_str).encode('utf-8') + msg = (random_str + url + (body_str or '')).encode('utf-8') hmac_obj = hmac.new(secret_key, digestmod=hashlib.sha256) hmac_obj.update(msg) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..d4839a6 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Tests package diff --git a/tests/test_iyzipay_resource.py b/tests/test_iyzipay_resource.py new file mode 100644 index 0000000..30ed641 --- /dev/null +++ b/tests/test_iyzipay_resource.py @@ -0,0 +1,107 @@ +import base64 +import unittest +from iyzipay.iyzipay_resource import IyzipayResource + + +class TestIyzipayResource(unittest.TestCase): + def setUp(self): + self.resource = IyzipayResource() + self.secret_key = "test_secret_key" + self.api_key = "test_api_key" + self.options = { + "api_key": self.api_key, + "secret_key": self.secret_key, + "base_url": "sandbox-api.iyzipay.com", + } + + def test_calculate_hmac_sha256_signature_with_strings(self): + params = ["12345", "TRY", "B67832", "conv123", "24.0", "20.0"] + sig = self.resource.calculate_hmac_sha256_signature(params, self.secret_key) + self.assertIsInstance(sig, str) + self.assertEqual(len(sig), 64) # SHA256 hex digest length + + def test_calculate_hmac_sha256_signature_with_non_string_types(self): + # Should gracefully handle numbers and None without raising TypeError + params = ["12345", None, 24.0, 1, 0] + sig = self.resource.calculate_hmac_sha256_signature(params, self.secret_key) + self.assertIsInstance(sig, str) + self.assertEqual(len(sig), 64) + + def test_verify_signature_returns_true_on_match(self): + params = ["payment123", "TRY", "100.0"] + sig = self.resource.calculate_hmac_sha256_signature(params, self.secret_key) + result = self.resource.verify_signature(params, self.secret_key, sig) + self.assertIs(result, True) + + def test_verify_signature_returns_false_on_mismatch(self): + params = ["payment123", "TRY", "100.0"] + tampered_sig = "a" * 64 + result = self.resource.verify_signature(params, self.secret_key, tampered_sig) + self.assertIs(result, False) + + def test_verify_signature_with_mixed_types(self): + params = ["payment123", None, 100.50] + sig = self.resource.calculate_hmac_sha256_signature(params, self.secret_key) + result = self.resource.verify_signature(params, self.secret_key, sig) + self.assertIs(result, True) + + def test_header_immutability_and_thread_safety(self): + # Verify that get_http_header returns independent copies + # and does not permanently mutate the class-level header + initial_keys = set(IyzipayResource.header.keys()) + + header1 = self.resource.get_http_header("/payment/test", self.options) + self.assertIn("Authorization", header1) + self.assertIn("x-iyzi-rnd", header1) + + # The class-level header must not be mutated + self.assertEqual(set(IyzipayResource.header.keys()), initial_keys) + self.assertNotIn("Authorization", IyzipayResource.header) + self.assertNotIn("x-iyzi-rnd", IyzipayResource.header) + + # Separate calls must return independent dictionaries + header2 = self.resource.get_http_header("/payment/test", self.options) + self.assertIsNot(header1, header2) + + def test_strip_zero(self): + self.assertEqual(IyzipayResource.strip_zero("10.0"), "10") + self.assertEqual(IyzipayResource.strip_zero("10.5"), "10.5") + self.assertEqual(IyzipayResource.strip_zero("10"), "10") + + def test_generate_random_string(self): + rnd = IyzipayResource.generate_random_string(8) + self.assertEqual(len(rnd), 8) + self.assertTrue(rnd.isalnum()) + + rnd16 = IyzipayResource.generate_random_string(16) + self.assertEqual(len(rnd16), 16) + + def test_generate_v2_hash(self): + v2_hash = IyzipayResource.generate_v2_hash( + self.api_key, + "/payment/test", + self.secret_key, + "random123", + '{"locale":"tr"}', + ) + self.assertIsInstance(v2_hash, str) + # Decodes valid base64 + decoded = base64.b64decode(v2_hash).decode("utf-8") + self.assertIn(f"apiKey:{self.api_key}", decoded) + self.assertIn("randomKey:random123", decoded) + self.assertIn("signature:", decoded) + + def test_generate_v2_hash_with_none_body(self): + # Handles None body without throwing TypeError + v2_hash = IyzipayResource.generate_v2_hash( + self.api_key, + "/payment/test", + self.secret_key, + "random123", + None, + ) + self.assertIsInstance(v2_hash, str) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pki_builder.py b/tests/test_pki_builder.py new file mode 100644 index 0000000..331d80b --- /dev/null +++ b/tests/test_pki_builder.py @@ -0,0 +1,45 @@ +import unittest +from iyzipay.pki_builder import PKIBuilder +from iyzipay.iyzipay_resource import IyzipayResource + + +class TestPKIBuilder(unittest.TestCase): + def test_pki_builder_append(self): + builder = PKIBuilder("") + builder.append("locale", "tr") + builder.append("conversationId", "123456789") + pki_str = builder.get_request_string() + self.assertEqual(pki_str, "[locale=tr,conversationId=123456789]") + + def test_pki_builder_append_price(self): + builder = PKIBuilder("") + builder.append_price("price", "10.50") + pki_str = builder.get_request_string() + self.assertEqual(pki_str, "[price=10.5]") + + def test_pki_builder_append_array(self): + builder = PKIBuilder("") + builder.append_array("items", ["[id=1]", "[id=2]"]) + pki_str = builder.get_request_string() + self.assertEqual(pki_str, "[items=[[id=1], [id=2]]]") + + def test_payment_card_pki(self): + card = { + "cardHolderName": "John Doe", + "cardNumber": "5528790000000008", + "expireYear": "2030", + "expireMonth": "12", + "cvc": "123", + "registerCard": "0", + "cardAlias": "my card", + "cardToken": "token123", + "cardUserKey": "userKey123", + } + pki_str = IyzipayResource.payment_card_pki(card) + self.assertIn("cardHolderName=John Doe", pki_str) + self.assertIn("cardNumber=5528790000000008", pki_str) + self.assertIn("cvc=123", pki_str) + + +if __name__ == "__main__": + unittest.main()