import binascii import datetime import os import time import typing import unicodedata import unittest import uuid from unittest import mock import bitcoin import rsa from bithustler_client.client.client import BithustlerClient from cards_client.client.client import CardsClient from cards_client.generated_protobuf import base_pb2 from conio_domain.wallet import WalletCryptoCurrency from vault_client.client.conio_user import ConioUser from vault_client.client.types import AcceptanceType, AcceptanceChoiceType from vault_client.generated_protobuf import vault_pb2 from conio_sdk.common import exceptions, ACCEPTANCE_TYPE_NAMES_BY_PB2 from conio_sdk.generated_protobuf import v1_pb2 from conio_sdk.services.conio.explicit_fees.explicit_fees_service import ExplicitFeesService from conio_sdk.services.conio.fiat_limits.fiat_limits_service import FiatLimitsService from conio_sdk.services.conio.sdk.crypto_proof.crypto_proof_verification_strategy import CryptoProofVerificationStrategy from conio_sdk.services.conio.sdk.signup_vo_service import SignupVOServices from conio_sdk.services.conio.user.user_service import UserService, AuthenticatedUser, AuthenticationData, \ UserCredentials, TermsAndConditionsAcceptances, UserLanguage from conio_sdk.services.conio.wallet.bitcoin_wallet_service import BitcoinWalletService, WalletEncryptedData _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 TestSignupVOServices(unittest.TestCase): def setUp(self): self.wallet_id = 'a wallet id' self.external_reference = 'an external reference' self.user_info = { 'reference_key_id': self.wallet_id, 'external_references': [self.external_reference], 'status': vault_pb2.ACTIVE, 'wallets': [ { 'reference_id': 'a wallet id', 'encrypted_private_key': os.urandom(16), 'encrypted_seed': os.urandom(16), 'encrypted_mnemonic': os.urandom(16), 'crypto_currency': WalletCryptoCurrency.BITCOIN } ], 'acceptances': { AcceptanceType.MARKETING: AcceptanceChoiceType.ACCEPTED, AcceptanceType.APP_IMPROVEMENT: AcceptanceChoiceType.REJECTED, AcceptanceType.CLIENT_SUPPORT: AcceptanceChoiceType.ACCEPTED } } self.user = ConioUser(data=self.user_info) self.pb2_user = vault_pb2.User( **{k: v for k, v in self.user_info.items() if k not in ('wallets', 'acceptances')}) self.pb2_user.wallets.add( reference_id=self.user.get_wallet(WalletCryptoCurrency.BITCOIN)['reference_id'], encrypted_private_key=self.user.get_wallet(WalletCryptoCurrency.BITCOIN)['encrypted_private_key'], encrypted_seed=self.user.get_wallet(WalletCryptoCurrency.BITCOIN)['encrypted_seed'], encrypted_mnemonic=self.user.get_wallet(WalletCryptoCurrency.BITCOIN)['encrypted_mnemonic'], ) for k, v in self.user.acceptances.items(): self.pb2_user.acceptances.add( acceptance_type=k.value, acceptance_choice=v.value ) self.user = ConioUser(data=self.user_info) self.user_service = mock.create_autospec(UserService) self.bitcoin_wallet_service = mock.create_autospec(BitcoinWalletService) self.fiat_limits_service = mock.create_autospec(FiatLimitsService) self.bithustler_client = mock.create_autospec(BithustlerClient) self.expiry_crypto_proof_verification_strategy = mock.create_autospec( CryptoProofVerificationStrategy ) self.signature_crypto_proof_verification_strategy = mock.create_autospec( CryptoProofVerificationStrategy ) self.explicit_fees_service = mock.create_autospec(ExplicitFeesService) self.cards_client = mock.create_autospec(CardsClient) self.cards_client.get_payment_methods.return_value = base_pb2.MsgGetPaymentMethodsResponse() self.sut = SignupVOServices( self.user_service, self.bitcoin_wallet_service, self.fiat_limits_service, self.explicit_fees_service, self.bithustler_client, self.cards_client, self.expiry_crypto_proof_verification_strategy, self.signature_crypto_proof_verification_strategy, ) def test_expired(self): self.expiry_crypto_proof_verification_strategy.verify.return_value = False with self.assertRaises(exceptions.CryptoProofExpiredException): self.sut.signup(mock.Mock(), UserLanguage('it')) def test_invalid_signature(self): self.signature_crypto_proof_verification_strategy.verify.return_value = False with self.assertRaises(exceptions.InvalidCryptoProofException): self.sut.signup(mock.Mock(), UserLanguage('it')) @classmethod def _create_signup_message( cls, first_name: typing.Optional[str] = None, last_name: typing.Optional[str] = None) -> 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.externalUserID = signup.conioCredentials.externalUserID signup.cryptoRequest.email = '{}@unit_tests.sdk.conio.com'.format(signup.cryptoRequest.externalUserID) if first_name: signup.cryptoRequest.firstName = first_name if last_name: signup.cryptoRequest.lastName = last_name data = '|'.join( filter( bool, ( signup.cryptoRequest.proofID, 'SIGNUP', signup.cryptoRequest.externalUserID, signup.cryptoRequest.userLevel, str(signup.cryptoRequest.proofExpiration), first_name, last_name ) ) ) data = unicodedata.normalize('NFC', data).encode('utf-8') signup.cryptoRequest.cryptoProof = rsa.pkcs1.sign(data, _HYPE_PRIVATE_KEY, 'SHA-256') return signup def test_signup(self): first_name = 'First name unit test {}'.format(datetime.datetime.now()) last_name = 'Last name unit test {}'.format(datetime.datetime.now()) authentication_data = AuthenticationData('an access token', 'a refresh token', 'a scope') authenticated_user = AuthenticatedUser(authentication_data, self.user) self.user_service.create_user_if_does_not_exist.return_value = self.user self.user_service.get_user.return_value = self.user self.user_service.activate_user_if_not_yet_active.return_value = authenticated_user msg = self._create_signup_message(first_name=first_name, last_name=last_name) user_level = msg.cryptoRequest.userLevel wallet_encrypted_data = WalletEncryptedData( encrypted_mnemonic=msg.encryptedUserKey.encryptedMnemonic, encrypted_seed=msg.encryptedUserKey.encryptedSeed, encrypted_private_key=msg.encryptedUserKey.encryptedPrivateKey, bip32_public_key=msg.encryptedUserKey.bip32PublicKey, wallet_id=None ) self.bitcoin_wallet_service.create_wallet_if_not_exists.return_value = wallet_encrypted_data signup_response, returned_authentication_data = self.sut.signup(msg, UserLanguage('it')) self.assertEqual(msg.tc, signup_response.tc) self.assertEqual(msg.encryptedUserKey, signup_response.encryptedUserKey) self.assertEqual(authentication_data, returned_authentication_data) self.user_service.create_user_if_does_not_exist.assert_called_once_with( UserCredentials( msg.conioCredentials.externalUserID, msg.conioCredentials.hashedConioPassword ), msg.cryptoRequest.email, first_name, last_name ) self.user_service.activate_user_if_not_yet_active.assert_called_once_with( UserCredentials( msg.conioCredentials.externalUserID, msg.conioCredentials.hashedConioPassword), TermsAndConditionsAcceptances( accepted_marketing=False, accepted_client_support=True, accepted_app_improvement=True ), 'it' ) self.bitcoin_wallet_service.create_wallet_if_not_exists.assert_called_once_with( authenticated_user.conio_user, wallet_encrypted_data ) self.explicit_fees_service.associate_explicit_fees.assert_called_once_with( authenticated_user.conio_user, user_level )