import typing from conio_domain.wallet import WalletCryptoCurrency from core_wallet_client.client.client import CoreWalletClient, FakeOutput from core_wallet_client.generated_protobuf import core_wallet_pb2 # noinspection PyPackageRequirements from core_wallet_client.generated_protobuf.core_wallet_pb2 import Signature from requests import HTTPError from vault_client.client.client_definition import VaultClient from vault_client.client.conio_user import ConioUser from vault_client.generated_protobuf import vault_pb2 from conio_sdk.common import exceptions from conio_sdk.common.decorators import xray_this from conio_sdk.logging.factory import LOGGING_FACTORY from conio_sdk.services.conio.wallet import bitcoin_wallet_service from conio_sdk.services.conio.wallet.bitcoin_wallet_service import WalletEncryptedData, BtcTransactionToBeSigned, \ BtcTransactionDataToSign, BTCAddressType class CoreWalletService(bitcoin_wallet_service.BitcoinWalletService): def __init__( self, core_wallet_client: CoreWalletClient, vault_client: VaultClient ): self._core_wallet_client = core_wallet_client self._vault_client = vault_client @xray_this def create_wallet_if_not_exists( self, user: ConioUser, wallet_encrypted_data: bitcoin_wallet_service.WalletEncryptedData)\ -> WalletEncryptedData: wallet = user.get_wallet(WalletCryptoCurrency.BITCOIN) if wallet: if any( map( lambda f: getattr(wallet_encrypted_data, f) != wallet[f], ('encrypted_seed', 'encrypted_private_key', 'encrypted_mnemonic') ) ): LOGGING_FACTORY.wallet.warning( 'A different wallet %s was already associated to user %s', wallet, user.reference_key_id) else: LOGGING_FACTORY.wallet.info( 'Wallet %s already associatesd to user %s', wallet, user.reference_key_id ) wallet_id = wallet['reference_id'] return bitcoin_wallet_service.WalletEncryptedData( encrypted_mnemonic=wallet['encrypted_mnemonic'], encrypted_seed=wallet['encrypted_seed'], encrypted_private_key=wallet['encrypted_private_key'], bip32_public_key=self.get_bip32_public_key(wallet_id), wallet_id=wallet_id ) else: wallet_id = self._create_wallet_if_not_exists(wallet_encrypted_data) return self._associate_wallet_to_user(user, wallet_encrypted_data, wallet_id) @xray_this def create_wallet( self, user: ConioUser, wallet_encrypted_data: bitcoin_wallet_service.WalletEncryptedData) \ -> WalletEncryptedData: wallet_id = self._core_wallet_client.create_wallet(wallet_encrypted_data.bip32_public_key) return self._associate_wallet_to_user(user, wallet_encrypted_data, wallet_id) @xray_this def get_bip32_public_key(self, bitcoin_wallet_id: bitcoin_wallet_service.WalletID) \ -> bitcoin_wallet_service.BIP32PublicKey: return bitcoin_wallet_service.BIP32PublicKey( self._core_wallet_client.get_signature_data(bitcoin_wallet_id).bip32_public_key ) @xray_this def get_wallet_balance(self, bitcoin_wallet_id: bitcoin_wallet_service.WalletID, invalidata_cache: bool = False) \ -> bitcoin_wallet_service.WalletBalance: wallet_info = self._core_wallet_client.get_wallet_info(bitcoin_wallet_id, invalidate_cache=invalidata_cache) return bitcoin_wallet_service.WalletBalance(wallet_info.balance, wallet_info.incoming_balance) @xray_this def get_current_btc_address(self, bitcoin_wallet_id: bitcoin_wallet_service.WalletID)\ -> bitcoin_wallet_service.BtcAddressID: return self._core_wallet_client.get_wallet_info(bitcoin_wallet_id).recv_address @xray_this def create_transaction( self, bitcoin_wallet_id: bitcoin_wallet_service.WalletID, btc_address: bitcoin_wallet_service.BtcAddressID, satoshi_amount: bitcoin_wallet_service.SatoshiAmount, fee_per_byte: typing.Optional[bitcoin_wallet_service.SatoshiAmount] = None, fake_outputs: typing.Sequence[FakeOutput] = ()) \ -> bitcoin_wallet_service.BtcTransactionToBeSigned: return self._create_btc_transaction_to_be_signed( self._core_wallet_client.create_transaction( bitcoin_wallet_id, btc_address, satoshi_amount, fee_per_byte=fee_per_byte, fake_outputs=fake_outputs ) ) @xray_this def create_all_transaction( self, bitcoin_wallet_id: bitcoin_wallet_service.WalletID, btc_address: bitcoin_wallet_service.BtcAddressID, fee_per_byte: typing.Optional[bitcoin_wallet_service.SatoshiAmount] = None, fake_outputs: typing.Sequence[FakeOutput] = ()) \ -> bitcoin_wallet_service.BtcTransactionToBeSigned: return self._create_btc_transaction_to_be_signed( self._core_wallet_client.create_all_transaction( bitcoin_wallet_id, btc_address, fee_per_byte=fee_per_byte, fake_outputs=fake_outputs ) ) @xray_this def submit_transaction( self, wallet_id: bitcoin_wallet_service.WalletID, tx: bitcoin_wallet_service.BtcRawTransaction, signatures: typing.List[bitcoin_wallet_service.BtcTransactionSignature] ) -> bitcoin_wallet_service.BtcTransactionHash: result = self._core_wallet_client.submit_transaction( wallet_id, tx, [Signature( bip44_path=s.bip44_path, signature=s.signature ) for s in signatures] ) if result.error.error_code != core_wallet_pb2.ERR_OK: raise exceptions.UnavailableBtcSubsystemException return result.data.tx_hash @xray_this def get_fees( self, wallet_id: bitcoin_wallet_service.WalletID, btc_addresses: typing.Sequence[bitcoin_wallet_service.BtcAddressID], fee_per_byte: int) -> int: # noinspection PyTypeChecker result = self._core_wallet_client.get_fees(wallet_id, btc_addresses, fee_per_byte) if result.error.error_code != core_wallet_pb2.ERR_OK: raise exceptions.UnavailableBtcSubsystemException return result.data.fees @xray_this def wallet_has_addresses(self, wallet_id: str) -> bool: return bool(self._core_wallet_client.get_wallet_addresses(wallet_id)) @classmethod def _create_btc_transaction_to_be_signed(cls, result: core_wallet_pb2.TransactionToBeSignedV1)\ -> bitcoin_wallet_service.BtcTransactionToBeSigned: return BtcTransactionToBeSigned( tx=result.tx, sign_data=[ BtcTransactionDataToSign( path=s.path, data_to_sign=s.data_to_sign, btc_address_type=BTCAddressType(s.address_type) ) for s in result.sign_data ], fees=result.fees, satoshi_amount=result.satoshi_amount, fee_per_byte=result.fee_per_byte if result.HasField('fee_per_byte') else None, locked_balance=result.locked_balance if result.HasField('locked_balance') else None, ) def _create_wallet_if_not_exists(self, wallet_encrypted_data) -> str: LOGGING_FACTORY.wallet.info('Creating wallet if not exists: %s', wallet_encrypted_data) try: existing_wallet_id = self._core_wallet_client.get_wallet_id_from_user_master_public_key( wallet_encrypted_data.bip32_public_key ) except HTTPError as e: if e.response.status_code == 404: LOGGING_FACTORY.wallet.info('Creating new wallet') existing_wallet_id = None else: LOGGING_FACTORY.wallet.exception('Error verifying existing wallet') raise e if existing_wallet_id: owning_user = self._vault_client.get_user_by_wallet_id(existing_wallet_id) if owning_user: LOGGING_FACTORY.wallet.error('Wallet %s already owned by user %s', existing_wallet_id, owning_user) raise exceptions.WalletAlreadyOwnedByAnotherUserException wallet_id = \ existing_wallet_id or \ self._core_wallet_client.create_wallet(wallet_encrypted_data.bip32_public_key) LOGGING_FACTORY.wallet.info('Returning wallet id %s', wallet_id) return wallet_id def _associate_wallet_to_user(self, user, wallet_encrypted_data, wallet_id): self._vault_client.save_user_info_v2( { 'wallets': [ { 'reference_id': wallet_id, 'encrypted_seed': wallet_encrypted_data.encrypted_seed, 'encrypted_private_key': wallet_encrypted_data.encrypted_private_key, 'encrypted_mnemonic': wallet_encrypted_data.encrypted_mnemonic, 'crypto_currency': vault_pb2.BITCOIN } ]}, user.reference_key_id) return bitcoin_wallet_service.WalletEncryptedData( encrypted_mnemonic=wallet_encrypted_data.encrypted_mnemonic, encrypted_seed=wallet_encrypted_data.encrypted_seed, encrypted_private_key=wallet_encrypted_data.encrypted_private_key, bip32_public_key=wallet_encrypted_data.bip32_public_key, wallet_id=bitcoin_wallet_service.WalletID(wallet_id) )