import os import time import typing import unicodedata import uuid import bitcoin import requests import rsa from aws_lambda_tools.common import openapi_json_serde from conio_sdk_generated.openapi.v2.serde.models.acceptance import Acceptance from conio_sdk_generated.openapi.v2.serde.models.acceptance_choice_type import AcceptanceChoiceType from conio_sdk_generated.openapi.v2.serde.models.acceptance_type import AcceptanceType from conio_sdk_generated.openapi.v2.serde.models.confirm_email import ConfirmEmail from conio_sdk_generated.openapi.v2.serde.models.conio_credentials import ConioCredentials from conio_sdk_generated.openapi.v2.serde.models.create_wallet import CreateWallet from conio_sdk_generated.openapi.v2.serde.models.encrypted_user_key import EncryptedUserKey from conio_sdk_generated.openapi.v2.serde.models.escalate_token import EscalateToken from conio_sdk_generated.openapi.v2.serde.models.full_signup import FullSignup from conio_sdk_generated.openapi.v2.serde.models.signup import Signup from conio_sdk_generated.openapi.v2.serde.models.signup_crypto_request import SignupCryptoRequest from conio_sdk_generated.openapi.v2.serde.models.success import Success from conio_sdk_generated.openapi.v2.serde.models.terms_and_conditions_acceptances import TermsAndConditionsAcceptances from requests import Response from test.integration import USER_LEVEL, ONE_MINUTE, HYPE_PRIVATE_KEY from test.integration.BaseIntegrationTest import BaseIntegrationTest class BaseV2IntegrationTest(BaseIntegrationTest): _URL_BASE = 'http://lambda-bff-sdk-v2_main:4001/development/v2/{}' def setUp(self): super(BaseV2IntegrationTest, self).setUp() @classmethod def create_full_signup_message( cls, expired: bool = False, first_name: typing.Optional[str] = None, last_name: typing.Optional[str] = None, email: typing.Optional[str] = None, external_user_id: typing.Optional[str] = None, hashed_conio_password: typing.Optional[str] = None ) -> FullSignup: seed = os.urandom(16) bip32_private_key = bitcoin.bip32_master_key(seed, vbytes=bitcoin.TESTNET_PRIVATE) user_id = external_user_id or str(uuid.uuid4()) signup = FullSignup( conio_credentials=ConioCredentials( user_id=user_id, password=hashed_conio_password or os.urandom(32).hex() ), tc=TermsAndConditionsAcceptances( acceptances=[ Acceptance( AcceptanceType.APP_IMPROVEMENT, AcceptanceChoiceType.ACCEPTED ), Acceptance( AcceptanceType.CLIENT_SUPPORT, AcceptanceChoiceType.ACCEPTED ) ] ), encrypted_user_key=EncryptedUserKey( # BEWARE: such data should be encrypted encrypted_mnemonic=b'encryptedMnemonic', encrypted_seed=b'encryptedSeed', encrypted_private_key=bip32_private_key.encode(), bip32_public_key=bitcoin.bip32_privtopub(bip32_private_key) ), crypto_request=SignupCryptoRequest( proof_id=str(uuid.uuid4()), proof_expiration=int(time.time() * 1000) + (-ONE_MINUTE if expired else ONE_MINUTE), iban=cls.random_iban(), email=email or cls._random_email(user_id), user_level=USER_LEVEL, first_name=first_name, last_name=last_name, external_user_id=user_id, crypto_proof=b'TO REPLACE' ) ) crypto_proof = '|'.join( filter( bool, ( signup.crypto_request.proof_id, 'SIGNUP', signup.crypto_request.external_user_id, signup.crypto_request.user_level, str(signup.crypto_request.proof_expiration), signup.crypto_request.iban, signup.crypto_request.email, signup.crypto_request.first_name, signup.crypto_request.last_name ) ) ) crypto_proof = unicodedata.normalize('NFC', crypto_proof).encode('utf-8') signup.crypto_request.crypto_proof = rsa.pkcs1.sign(crypto_proof, HYPE_PRIVATE_KEY, 'SHA-256') return signup def post( self, path: str, msg, response_type: typing.Optional[typing.Type] = None, auth_token: typing.Optional[str] = None, **kwargs ) -> typing.Union[object, Response]: kwargs = dict(kwargs) kwargs['headers'] = dict(kwargs.get('headers', dict())) if 'Content-Type' not in kwargs['headers']: kwargs['headers']['Content-Type'] = 'application/json' if auth_token: kwargs['headers']['Authorization'] = f'Bearer {auth_token}' response = requests.post( url=self._URL_BASE.format(path), data=openapi_json_serde.serialize(msg), **kwargs ) response.raise_for_status() if response_type: return openapi_json_serde.deserialize(response.content, response_type) return response def form_post( self, path: str, payload: dict, response_type: typing.Optional[typing.Type] = None, auth_token: typing.Optional[str] = None, **kwargs ) -> typing.Union[object, Response]: kwargs = dict(kwargs) kwargs['headers'] = dict(kwargs.get('headers', dict())) if 'Content-Type' not in kwargs['headers']: kwargs['headers']['Content-Type'] = 'application/x-www-form-urlencoded' if auth_token: kwargs['headers']['Authorization'] = f'Bearer {auth_token}' response = requests.post( url=self._URL_BASE.format(path), data=payload, **kwargs ) response.raise_for_status() if response_type: return openapi_json_serde.deserialize(response.content, response_type) return response @classmethod def create_create_wallet_message(cls) -> CreateWallet: seed = os.urandom(32) bip32_private_key = bitcoin.bip32_master_key(seed, vbytes=bitcoin.TESTNET_PRIVATE) return CreateWallet( encrypted_user_key=EncryptedUserKey( encrypted_mnemonic=os.urandom(32), encrypted_seed=seed[::-1], encrypted_private_key=bip32_private_key.encode()[::-1], bip32_public_key=bitcoin.bip32_privtopub(bip32_private_key) ) ) @classmethod def create_signup_message(cls) -> Signup: return Signup( conio_credentials=ConioCredentials( user_id=cls._random_email(), password=str(uuid.uuid4()) ), device_id=str(uuid.uuid4()), acceptances=[ Acceptance( AcceptanceType.APP_IMPROVEMENT, AcceptanceChoiceType.ACCEPTED ), Acceptance( AcceptanceType.CLIENT_SUPPORT, AcceptanceChoiceType.ACCEPTED ) ] ) @classmethod def _random_email(cls, username: typing.Optional[str] = None) -> str: return '{}@test_bff_sdk_v2.conio.com'.format(username or uuid.uuid4()) def confirm_email(self, email: str): self.post('test/confirm_email', ConfirmEmail(email=email), response_type=Success) def escalate_token(self, access_token: str, physical_device_id: str) -> Response: response = self.post('escalate_token', EscalateToken(physical_device_id=physical_device_id), auth_token=access_token) self.assertEqual('success', openapi_json_serde.deserialize(response.content, Success).status) return response def login_email( self, username: str, password: str, hardware_id: str, scope: str) -> Response: return self.form_post( 'login_email', payload={'username': username, 'password': password, 'scope': scope, 'grant_type': 'password'}, headers={'X-ConioHardwareID': hardware_id}) def login_external_user_id( self, username: str, password: str, hardware_id: str, scope: str) -> Response: return self.form_post( 'login_external_user_id', payload={'grant_type': 'password', 'username': username, 'password': password, 'scope': scope}, headers={'X-ConioHardwareID': hardware_id})