import binascii import itertools import os import time import unicodedata import unittest import uuid import bitcoin import rsa from pycomb.exceptions import PyCombValidationError from conio_sdk.combinators import v1 from conio_sdk.common import ACCEPTANCE_TYPE_NAMES_BY_PB2 from conio_sdk.generated_protobuf import v1_pb2 from conio_sdk.combinators.v1 import COMBINATORS _HYPE_PRIVATE_KEY_DATA = b''' -----BEGIN RSA PRIVATE KEY----- MIIEogIBAAKCAQEA1aYX3mtodPYVIPO8Vb2czaDEJhl3O8xaupLnA9tOeRnDrrTb o5G8CnxSXXC9FssEcG/CNYyQY5Df8n+KkuATfZA/rbMUlvOVtOgvPJuzKL9SZCai 2V0V/qWlK8fuZWvKQzvtUqE/2y9cq+aCUHbrGj8qSo9eALNmz4s3RvV9GFhhx8uE mTYHf3Pm89fbpoPCOpKGHEBLoqPkcKALl2zsgRwZqgG3tARKymcXpjhDwTtV5c+A rSBBV3OFWZ6nxMe1GgXQZWuEi4JmO93eDQujyc/TfmqB1Q+VoXeqDwmBECjNOYd2 CueRmtwsILPBLoLhGkE+YpAEEPM4QgI7xb7zTwIDAQABAoIBAHaA1RZynxLY9+k6 KEmqjZHkzUeQsnkBpYV9PBQAjatQJiD+giFdEV8DjC/1+3vsCb9PzfojyGbhkcYR BkznawgnfZqcDRyZaX1Zl/HXLu24CTwxzfwgzLVdLZt2Hv40ZpEaaU1+0UuDHrTe e4OkIk2BobSPhwV+fNU7k+KRAd0BDTfzp7Vt1m0MCMNU55qUxo5KlNIUtv39Oxsm wx8v4r+cZIIfJEeW0xSQHa5ZN8dwPgXbB1p97RhDl2j7UAnMM63smvQJ40MMcCx7 m6plzv4wd7+k384Sju5pjzO0pgW/Ne4XtCu6rqNWrI2miZ1xyH2JuQR73F0KulKX jNusFjECgYEA8+vvRNGTn+8bFwnq3aLqtlzPnqe6pcpUGhTeHzt/g9izYOIO3t6D hYZlJbN4n3C6RkcJ4wMFgN6SbW+nzw09dca6j465BkXfffkt4g8kWeCl5a0hbY0z /JIFzTo8X32sIvboVvjb61nDoW6egdCufuN28Ib9f1M7PeMqiVJfTIcCgYEA4Dpn 9ahQEoNWFowEXdghEfxU5nqIVJr6OVvZmRqH+haCe3Oo+Ap9WasdXBFOdjePRPkc sbtV3K0ER59Z+dRrXrLWOtOw9LSOmGSFWSuepKN/JJ0wJyBssYMM1ZqVGnTg/nqO Ja/1lwOnxZbezrrQtwm8+6K7zWt58EMnDiRLXPkCgYArknrUZUekqzbAn9Hns6GP 3/ZqlfW+he0OF6oyFBPMPpqUdO1JHKCL6p0I5g1nFeEAitIWTkTeZ2PqzqZAU1Im RtCuskUU/MhWnXt3xVKuB3Y7F/k/s5iUxpTouz1rpWxpdoe8eYn3ebp7jOIduGRj YEiv4L1J0Fllzb2ceC1z4wKBgAT9B6cNcYqX5WhnAQndbw7pYDIoc7P+Jqb0BilD z9aefZSlhBLQmO1Pwz1zHR3AKq3MJPlHQ6e/KaM2Rlgqg6D9tYplf0BSbAGz6suL DuJ2yLNV0+Zq8EAavERcRgjqpL7Elzj7aylK6YaZzqcmvNH1o4CtpCPzyiiwNcQ4 xnxxAoGAGC+baVi02c9ZjS3/Le8MPL12/r8CJrMYYLAONuijspsNDDke8cs3Jl4L 5uTiXWW+6kBo0rgfbF6k09IkcYkcOu5fwl/cvXZw64Z+H4g7bcbQv+OcKQx1/nsf WqOOUzGI+AUZX3MS1SalCbf2r5zajezR+pLYqGQMBV4Ix77A6Lw= -----END RSA PRIVATE KEY----- ''' _HYPE_PRIVATE_KEY = rsa.PrivateKey.load_pkcs1(_HYPE_PRIVATE_KEY_DATA) # type: rsa.PublicKey class TestCombinators(unittest.TestCase): _FUNCTION_TYPE = type(lambda: None) @classmethod def create_signup_message(cls) -> v1_pb2.Signup: signup = v1_pb2.Signup() signup.conioCredentials.externalUserID = str(uuid.uuid4()) signup.conioCredentials.hashedConioPassword = binascii.hexlify(os.urandom(32)).decode() signup.tc.acceptances.add( acceptance_type=ACCEPTANCE_TYPE_NAMES_BY_PB2[v1_pb2.ACCEPTANCE_CLIENT_SUPPORT], acceptance_choice=v1_pb2.ACC_CHOICE_ACCEPTED ) signup.tc.acceptances.add( acceptance_type=ACCEPTANCE_TYPE_NAMES_BY_PB2[v1_pb2.ACCEPTANCE_APP_IMPROVEMENT], acceptance_choice=v1_pb2.ACC_CHOICE_ACCEPTED ) seed = os.urandom(16) bip32_private_key = bitcoin.bip32_master_key(seed, vbytes=bitcoin.TESTNET_PRIVATE) signup.encryptedUserKey.encryptedMnemonic = b'encryptedMnemonic' # BEWARE: such data should be encrypted signup.encryptedUserKey.encryptedSeed = b'encryptedSeed' signup.encryptedUserKey.encryptedPrivateKey = bip32_private_key.encode() signup.encryptedUserKey.bip32PublicKey = bitcoin.bip32_privtopub(bip32_private_key) signup.cryptoRequest.proofID = str(uuid.uuid4()) signup.cryptoRequest.proofExpiration = int(time.time() * 1000) + 1000 signup.cryptoRequest.userLevel = 'A Smart Level' signup.cryptoRequest.iban = 'IT60X0542811101000000123456' signup.cryptoRequest.email = '3aa6d1af-a5be-420b-b814-7314be1cbd9c@test_lambda_bff_sdk.conio.com' signup.cryptoRequest.externalUserID = signup.conioCredentials.externalUserID data = '|'.join( ( signup.cryptoRequest.proofID, 'SIGNUP', signup.cryptoRequest.externalUserID, signup.cryptoRequest.userLevel, str(signup.cryptoRequest.proofExpiration), signup.cryptoRequest.iban ) ) data = unicodedata.normalize('NFC', data).encode('utf-8') signup.cryptoRequest.cryptoProof = rsa.pkcs1.sign(data, _HYPE_PRIVATE_KEY, 'SHA-256') return signup @classmethod def _verify(cls, msg): COMBINATORS[type(msg)](msg) def test_signup(self): msg = self.create_signup_message() self._verify(msg) msg.encryptedUserKey.bip32PublicKey = 'hello' with self.assertRaises(PyCombValidationError): self._verify(msg) msg = self.create_signup_message() msg.tc.acceptances[0].acceptance_type = '' with self.assertRaises(PyCombValidationError): self._verify(msg) def test_login(self): msg = v1_pb2.Login() msg.conioCredentials.externalUserID = 'a user id' msg.conioCredentials.hashedConioPassword = 'ab1234ef' self._verify(msg) msg.conioCredentials.hashedConioPassword = 'hello' with self.assertRaises(PyCombValidationError): self._verify(msg) def test_full_coverage(self): for c in v1.COMBINATORS.keys(): print(c) self.assertTrue( c in v1.COMBINATORS, msg='Class {} not protected by combinator'.format(c)) self._recursively_check_combinators( c.DESCRIPTOR, v1.COMBINATORS[c], c.__name__ ) def _recursively_check_combinators(self, proto_msg_descriptor, proto_msg_combinator, path): proto_msg_type = self._get_class_from_descriptor(proto_msg_descriptor) for cur_proto_descriptor in proto_msg_descriptor.fields: cur_path = '{}.{}'.format(path, cur_proto_descriptor.name) cur_proto_msg_combinators = self._get_subvalidators( proto_msg_type, proto_msg_combinator, cur_proto_descriptor, cur_path) for cur_proto_msg_combinator in cur_proto_msg_combinators: self.assertIsNotNone(cur_proto_msg_combinator, msg='Object {}.{} has no combinator'.format( proto_msg_type, cur_path)) if cur_proto_descriptor.type == 11: self._recursively_check_combinators( cur_proto_descriptor.message_type, cur_proto_msg_combinator, path=cur_path) @classmethod def _get_class_from_descriptor(cls, descriptor) -> type: real_descriptor = descriptor if type(descriptor).__name__ in ('MessageDescriptor', 'Descriptor') else descriptor.message_type return getattr(v1_pb2, real_descriptor.full_name.replace('conio.sdk.v1.', '')) def _get_subvalidators(self, proto_msg_type, proto_msg_combinator, cur_proto_descriptor, cur_path) -> list: try: # This is subtype, or a maybe, or a union, or... well, not a generic object. That sucks... if not proto_msg_combinator.__closure__: print('proto_msg_type: {}'.format(proto_msg_type)) print('proto_msg_combinator: {} | {}'.format(proto_msg_combinator, dir(proto_msg_combinator))) if type(proto_msg_combinator.__closure__[0].cell_contents) == self._FUNCTION_TYPE: # Oh god, a list... # if str(proto_msg_combinator.__closure__[0].cell_contents.meta).startswith('{\'name\': \'List('): # pass # Union. Ugh... if proto_msg_combinator.__closure__[0].cell_contents.__name__ == '_union': union_combinators = proto_msg_combinator.__closure__[1].cell_contents return list(itertools.chain( self._get_subvalidators( proto_msg_type, x, cur_proto_descriptor, cur_path ) for x in union_combinators) ) if str(proto_msg_combinator.__closure__[0].cell_contents.meta).startswith('{\'name\': \'Constant('): return [] # This is a subtype return self._get_subvalidators( proto_msg_type, proto_msg_combinator.__closure__[0].cell_contents, cur_proto_descriptor, cur_path) except AttributeError as e: raise e try: cur_proto_msg_combinator = proto_msg_combinator.__closure__[0].cell_contents[cur_proto_descriptor.name] except KeyError: self.fail('No such combinator for field {}'.format(cur_path)) return [cur_proto_msg_combinator] def test_history_prices(self): msg = v1_pb2.MsgHistoryPrices( currency=v1_pb2.EUR, end_timestamp=1555421473000, interval=86400, start_timestamp=1523885446000 ) d = msg.SerializeToString() msg2 = v1_pb2.MsgHistoryPrices() msg2.ParseFromString(d) print(str(msg)) print(str(msg2)) with self.assertRaises(PyCombValidationError): self._verify(msg) def test_create_ask(self): msg = v1_pb2.MsgCreateOrRefreshAsk( currency=v1_pb2.EUR ) self._verify(msg) msg.fiat_amount = 12345 self._verify(msg) msg = v1_pb2.MsgCreateOrRefreshAsk( currency=v1_pb2.EUR ) self._verify(msg) msg.satoshi = 12345 self._verify(msg) def test_create_bid(self): msg = v1_pb2.MsgCreateOrRefreshBid( currency=v1_pb2.EUR ) with self.assertRaises(PyCombValidationError): self._verify(msg) msg.fiat_amount = 12345 self._verify(msg) msg = v1_pb2.MsgCreateOrRefreshBid( currency=v1_pb2.EUR ) with self.assertRaises(PyCombValidationError): self._verify(msg) msg.satoshi = 12345 self._verify(msg)