diff --git a/blocktrail/__init__.py b/blocktrail/__init__.py index 769a344..4bafeed 100644 --- a/blocktrail/__init__.py +++ b/blocktrail/__init__.py @@ -19,3 +19,4 @@ def to_btc(satoshi): from blocktrail import connection from blocktrail import exceptions from blocktrail.client import APIClient +from blocktrail.wallet import Wallet diff --git a/blocktrail/client.py b/blocktrail/client.py index 1c95432..39f0a09 100644 --- a/blocktrail/client.py +++ b/blocktrail/client.py @@ -1,4 +1,10 @@ +import os +from bitcoin import SelectParams +from bitcoin.wallet import CBitcoinSecret, P2PKHBitcoinAddress from blocktrail import connection +from blocktrail.wallet import Wallet +from mnemonic.mnemonic import Mnemonic +from pycoin.key.BIP32Node import BIP32Node class APIClient(object): @@ -14,9 +20,14 @@ def __init__(self, api_key, api_secret, network='BTC', testnet=False, api_versio :param bool debug: print debug information when requests fail """ + self.testnet = testnet + + SelectParams('testnet' if self.testnet else 'mainnet') + if api_endpoint is None: network = ("t" if testnet else "") + network.upper() - api_endpoint = "https://api.blocktrail.com/%s/%s" % (api_version, network) + api_endpoint = os.environ.get('BLOCKTRAIL_SDK_API_ENDPOINT', "https://api.blocktrail.com") + api_endpoint = "%s/%s/%s" % (api_endpoint, api_version, network) self.client = connection.RestClient(api_endpoint=api_endpoint, api_key=api_key, api_secret=api_secret, debug=debug) @@ -330,3 +341,199 @@ def verify_message(self, message, address, signature): )) return response.json()['result'] + + def all_wallets(self, page=1, limit=20): + """ + get all wallets (paginated) + + :param int page: pagination page, starting at 1 + :param int limit: the amount of wallets per page, can be between 1 and 200 + :rtype: dict + """ + + response = self.client.get("/wallets", params={'page': page, 'limit': limit}, auth=True) + + return response.json() + + def create_new_wallet(self, identifier, passphrase, key_index=0): + netcode = "XTN" if self.testnet else "BTC" + + primary_mnemonic = Mnemonic(language='english').generate(strength=512) + primary_seed = Mnemonic.to_seed(primary_mnemonic, passphrase) + primary_private_key = BIP32Node.from_master_secret(primary_seed, netcode=netcode) + + primary_public_key = primary_private_key.subkey_for_path("%d'.pub" % key_index) + + backup_mnemonic = Mnemonic(language='english').generate(strength=512) + backup_seed = Mnemonic.to_seed(backup_mnemonic, "") + backup_public_key = BIP32Node.from_master_secret(backup_seed, netcode=netcode).public_copy() + + checksum = self.create_checksum(primary_private_key) + + result = self._create_new_wallet( + identifier=identifier, + primary_public_key=(primary_public_key.as_text(), "M/%d'" % key_index), + backup_public_key=(backup_public_key.as_text(), "M"), + primary_mnemonic=primary_mnemonic, + checksum=checksum, + key_index=key_index + ) + + blocktrail_public_keys = result['blocktrail_public_keys'] + key_index = result['key_index'] + + return Wallet( + client=self, + identifier=identifier, + primary_mnemonic=primary_mnemonic, + primary_private_key=primary_private_key, + backup_public_key=backup_public_key, + blocktrail_public_keys=blocktrail_public_keys, + key_index=key_index, + testnet=self.testnet + ), primary_mnemonic, backup_mnemonic, blocktrail_public_keys + + def _create_new_wallet(self, identifier, primary_public_key, backup_public_key, primary_mnemonic, checksum, key_index): + response = self.client.post("/wallet", data={ + 'identifier': identifier, + 'primary_public_key': primary_public_key, + 'backup_public_key': backup_public_key, + 'primary_mnemonic': primary_mnemonic, + 'checksum': checksum, + 'key_index': key_index, + }, auth=True) + + return response.json() + + def init_wallet(self, identifier, passphrase): + netcode = "XTN" if self.testnet else "BTC" + + data = self.get_wallet(identifier) + + primary_seed = Mnemonic.to_seed(data['primary_mnemonic'], passphrase) + primary_private_key = BIP32Node.from_master_secret(primary_seed, netcode=netcode) + + backup_public_key = BIP32Node.from_hwif(data['backup_public_key'][0]) + + checksum = self.create_checksum(primary_private_key) + if checksum != data['checksum']: + raise Exception("Checksum [%s] does not match expected checksum [%s], most likely due to incorrect password" % (checksum, data['checksum'])) + + blocktrail_public_keys = data['blocktrail_public_keys'] + key_index = data['key_index'] + + return Wallet( + client=self, + identifier=identifier, + primary_mnemonic=data['primary_mnemonic'], + primary_private_key=primary_private_key, + backup_public_key=backup_public_key, + blocktrail_public_keys=blocktrail_public_keys, + key_index=key_index, + testnet=self.testnet + ) + + @staticmethod + def create_checksum(key): + key = CBitcoinSecret(key.wif()) + address = P2PKHBitcoinAddress.from_pubkey(key.pub) + + return str(address) + + def get_wallet(self, identifier): + response = self.client.get("/wallet/%s" % (identifier, ), auth=True) + + return response.json() + + def get_wallet_balance(self, identifier): + response = self.client.get("/wallet/%s/balance" % (identifier, ), auth=True) + + return response.json() + + def wallet_discovery(self, identifier, gap=200): + # @TODO: 360s timeout + response = self.client.get("/wallet/%s/discovery" % (identifier, ), params=dict(gap=gap), auth=True) + + return response.json() + + def get_new_derivation(self, identifier, path): + response = self.client.post("/wallet/%s/path" % (identifier, ), data={ + 'path': path, + }, auth=True) + + return response.json() + + def upgrade_key_index(self, identifier, key_index, primary_public_key): + response = self.client.post( + "/wallet/%s/upgrade" % (identifier, ), + data=dict( + key_index=key_index, + primary_public_key=primary_public_key + ), + auth=True + ) + + return response.json() + + def coin_selection(self, identifier, outputs, lockUTXO=False, allow_zero_conf=False): + response = self.client.post( + "/wallet/%s/coin-selection" % (identifier, ), + params={ + 'lock': lockUTXO, + 'zeroconf': allow_zero_conf + }, + data=outputs, + auth=True + ) + + return response.json() + + def send_transaction(self, identifier, raw_tx, paths, check_fee=False): + response = self.client.post( + "/wallet/%s/send" % (identifier, ), + params={ + 'check_fee': check_fee + }, + data={ + 'raw_transaction': raw_tx, + 'paths': paths + }, + auth=True + ) + + return response.json() + + def wallet_transactions(self, identifier, page=1, limit=20): + response = self.client.get("/wallet/%s/transactions" % (identifier, ), params={'page': page, 'limit': limit}, auth=True) + + return response.json() + + def wallet_addresses(self, identifier, page=1, limit=20): + response = self.client.get("/wallet/%s/addresses" % (identifier, ), params={'page': page, 'limit': limit}, auth=True) + + return response.json() + + def setup_wallet_webhook(self, wallet_identifier, webhook_identifier, url): + """ + create a new webhook for a wallet + + :param str wallet_identifier: the wallet identifier which which to create te webhook + :param str webhook_identifier: a unique identifier to associate with this webhook + :param str url: the url to receive the webhook events + :rtype: dict + """ + response = self.client.post("/wallet/%s/webhook" % (wallet_identifier, ), data={'url': url, 'identifier': webhook_identifier}, auth=True) + + return response.json() + + def delete_wallet_webhook(self, wallet_identifier, webhook_identifier): + """ + deletes an existing webhook for a wallet + + :param str wallet_identifier: the wallet identifier which which to create te webhook + :param str webhook_identifier: a unique identifier to associate with this webhook + :rtype: dict + """ + response = self.client.delete("/wallet/%s/webhook/%s" % (wallet_identifier, webhook_identifier, ), auth=True) + + return response.json() diff --git a/blocktrail/connection.py b/blocktrail/connection.py index f82d796..ea8c99c 100644 --- a/blocktrail/connection.py +++ b/blocktrail/connection.py @@ -8,7 +8,11 @@ import json import hashlib -from httpsig.requests_auth import HTTPSignatureAuth +try: + from httpsig_cffi.requests_auth import HTTPSignatureAuth +except: + from httpsig.requests_auth import HTTPSignatureAuth + from requests.models import RequestEncodingMixin import blocktrail @@ -56,17 +60,24 @@ def get(self, endpoint_url, params=None, auth=None): :param bool auth: do HMAC auth :rtype: requests.Response """ + endpoint_url = self.api_endpoint + endpoint_url + if auth is True: auth = self.auth + params = dict_merge(self.default_params, params) + + params['api_kdy'] = 'ruben1' + params['api_kfy'] = 'ruben1' + + params = RestClient.sort_params(params) + headers = dict_merge(self.default_headers, { 'Date': RestClient.httpdate(datetime.datetime.utcnow()), - 'Content-MD5': RestClient.content_md5("") + 'Content-MD5': RestClient.content_md5(urlparse(endpoint_url).path + "?" + urlencode(params)) }) - params = dict_merge(self.default_params, params) - - response = requests.get(self.api_endpoint + endpoint_url, params=params, headers=headers, auth=auth) + response = requests.get(endpoint_url, params=params, headers=headers, auth=auth) return self.handle_response(response) @@ -77,9 +88,14 @@ def post(self, endpoint_url, data, params=None, auth=None): :param bool auth: do HMAC auth :rtype: requests.Response """ + endpoint_url = self.api_endpoint + endpoint_url + if auth is True: auth = self.auth + params = dict_merge(self.default_params, params) + params = RestClient.sort_params(params) + # do the post body encoding here since we need it to get the MD5 data = json.dumps(data) @@ -89,8 +105,7 @@ def post(self, endpoint_url, data, params=None, auth=None): 'Content-Type': 'application/json' }) - params = dict_merge(self.default_params, params) - response = requests.post(self.api_endpoint + endpoint_url, data=data, params=params, headers=headers, auth=auth) + response = requests.post(endpoint_url, data=data, params=params, headers=headers, auth=auth) return self.handle_response(response) @@ -101,9 +116,14 @@ def put(self, endpoint_url, data, params=None, auth=None): :param bool auth: do HMAC auth :rtype: requests.Response """ + endpoint_url = self.api_endpoint + endpoint_url + if auth is True: auth = self.auth + params = dict_merge(self.default_params, params) + params = RestClient.sort_params(params) + # do the post body encoding here since we need it to get the MD5 data = json.dumps(data) @@ -113,8 +133,7 @@ def put(self, endpoint_url, data, params=None, auth=None): 'Content-Type': 'application/json' }) - params = dict_merge(self.default_params, params) - response = requests.put(self.api_endpoint + endpoint_url, data=data, params=params, headers=headers, auth=auth) + response = requests.put(endpoint_url, data=data, params=params, headers=headers, auth=auth) return self.handle_response(response) @@ -125,20 +144,28 @@ def delete(self, endpoint_url, data=None, params=None, auth=None): :param bool auth: do HMAC auth :rtype: requests.Response """ + endpoint_url = self.api_endpoint + endpoint_url + if auth is True: auth = self.auth - data = json.dumps(data) - params = dict_merge(self.default_params, params) + params = RestClient.sort_params(params) + + if data: + # do the post body encoding here since we need it to get the MD5 + data = json.dumps(data) + content_md5 = RestClient.content_md5(data) + else: + content_md5 = RestClient.content_md5(urlparse(endpoint_url).path + "?" + urlencode(params)) headers = dict_merge(self.default_headers, { 'Date': RestClient.httpdate(datetime.datetime.utcnow()), - 'Content-MD5': RestClient.content_md5(urlparse(self.api_endpoint + endpoint_url).path + "?" + urlencode(params)), + 'Content-MD5': content_md5, 'Content-Type': 'application/json' }) - response = requests.delete(self.api_endpoint + endpoint_url, data=data, params=params, headers=headers, auth=auth) + response = requests.delete(endpoint_url, data=data, params=params, headers=headers, auth=auth) return self.handle_response(response) @@ -157,24 +184,28 @@ def handle_response(self, response): elif self.debug: print(response.url, response.status_code, response.content) - if response.status_code == 400 or response.status_code == 403: + data = {} + try: data = response.json() + except Exception: + pass + if response.status_code == 400 or response.status_code == 403: if data and data['msg'] and data['code']: raise EndpointSpecificError(msg=data['msg'], code=data['code']) else: raise UnknownEndpointSpecificError(EXCEPTION_UNKNOWN_ENDPOINT_SPECIFIC_ERROR) elif response.status_code == 401: - raise InvalidCredentials(msg=EXCEPTION_INVALID_CREDENTIALS, code=401) + raise InvalidCredentials(msg=data.get('msg', EXCEPTION_INVALID_CREDENTIALS), code=401) elif response.status_code == 404: if response.reason == "Endpoint Not Found": - raise MissingEndpoint(msg=EXCEPTION_MISSING_ENDPOINT, code=404) + raise MissingEndpoint(msg=data.get('msg', EXCEPTION_MISSING_ENDPOINT), code=404) else: - raise ObjectNotFound(msg=EXCEPTION_OBJECT_NOT_FOUND, code=404) + raise ObjectNotFound(msg=data.get('msg', EXCEPTION_OBJECT_NOT_FOUND), code=404) elif response.status_code == 500: - raise GenericServerError(msg=EXCEPTION_GENERIC_SERVER_ERROR, code=response.status_code) + raise GenericServerError(msg=data.get('msg', EXCEPTION_GENERIC_SERVER_ERROR), code=response.status_code) else: - raise GenericHTTPError(msg=EXCEPTION_GENERIC_HTTP_ERROR, code=response.status_code) + raise GenericHTTPError(msg=data.get('msg', EXCEPTION_GENERIC_HTTP_ERROR), code=response.status_code) @classmethod def content_md5(cls, content=""): @@ -192,6 +223,10 @@ def httpdate(cls, dt): "Oct", "Nov", "Dec"][dt.month - 1] return "%s, %02d %s %04d %02d:%02d:%02d GMT" % (weekday, dt.day, month, dt.year, dt.hour, dt.minute, dt.second) + @classmethod + def sort_params(cls, params): + return sorted([(k, v) for k, v in params.items()], key=lambda t: t[0]) + def dict_merge(dict1, dict2): dict1 = dict1 if dict1 is not None else {} diff --git a/blocktrail/wallet.py b/blocktrail/wallet.py new file mode 100644 index 0000000..ba6e6b5 --- /dev/null +++ b/blocktrail/wallet.py @@ -0,0 +1,177 @@ +import struct +import random +from pycoin.key.BIP32Node import BIP32Node +from bitcoin.core import x, b2x, lx, COutPoint, CMutableTxOut, CMutableTxIn, CMutableTransaction +from bitcoin.core.script import CScript, SignatureHash, SIGHASH_ALL, OP_CHECKMULTISIG, OP_0 +from bitcoin.wallet import CBitcoinAddress, CBitcoinSecret, P2PKHBitcoinAddress + +VERIFY_NEW_DERIVATIONS = True + + +class Wallet(object): + def __init__(self, client, identifier, primary_mnemonic, primary_private_key, backup_public_key, blocktrail_public_keys, key_index, testnet): + """ + @type primary_private_key: BIP32Node + @type backup_public_key: BIP32Node + """ + self.client = client + self.identifier = identifier + self.primary_mnemonic = primary_mnemonic + self.primary_private_key = primary_private_key + self.backup_public_key = backup_public_key + + self.blocktrail_public_keys = dict([(str(_key_index), BIP32Node.from_hwif(_key[0])) for _key_index, _key in blocktrail_public_keys.items()]) + self.key_index = int(key_index) + self.testnet = testnet + + def get_new_address_pair(self): + path = self.get_new_derivation() + address = self.get_address_by_path(path) + + return path, address + + def get_new_derivation(self): + parent_path = "M/%d'/0" % self.key_index + result = self.client.get_new_derivation(self.identifier, path=parent_path) + + path = result['path'] + + if VERIFY_NEW_DERIVATIONS: + if result['address'] != self.get_address_by_path(path): + raise Exception("Failed to verify that address from API matches address locally") + + return path + + def get_address_by_path(self, path, key=None): + path = path.replace("M/", "") + key_index = path.split("/")[0].replace("'", "") + + if key is None: + key = self.primary_private_key.subkey_for_path(path) + + backup_public_key = self.backup_public_key.subkey_for_path(path.replace("'", "")) + blocktrail_public_key = self.blocktrail_public_keys[str(key_index)].subkey_for_path("/".join(path.split("/")[1:])) + + redeemScript = CScript([2] + sorted([ + key.sec(use_uncompressed=False), + backup_public_key.sec(use_uncompressed=False), + blocktrail_public_key.sec(use_uncompressed=False), + ]) + [3, OP_CHECKMULTISIG]) + + scriptPubKey = redeemScript.to_p2sh_scriptPubKey() + address = CBitcoinAddress.from_scriptPubKey(scriptPubKey) + + return str(address) + + def get_balance(self): + balance_info = self.client.get_wallet_balance(self.identifier) + + return balance_info['confirmed'], balance_info['unconfirmed'] + + def pay(self, pay, change_address=None, allow_zero_conf=False, randomize_change_idx=True): + send = {} + + if isinstance(pay, list): + for address, value in pay: + send[address] = value + else: + send = pay + + coin_selection = self.client.coin_selection(self.identifier, send, lockUTXO=True, allow_zero_conf=allow_zero_conf) + + utxos = coin_selection['utxos'] + fee = coin_selection['fee'] + change = coin_selection['change'] + + if change > 0: + if change_address is None: + _, change_address = self.get_new_address_pair() + + send[change_address] = change + + txins = [] + for utxo in utxos: + txins.append(CMutableTxIn(COutPoint(lx(utxo['hash']), utxo['idx']))) + + txouts = [] + change_txout = None + for address, value in send.items(): + txout = CMutableTxOut(value, CBitcoinAddress(address).to_scriptPubKey()) + if address == change_address: + change_txout = txout + txouts.append(txout) + + # randomly move the change_txout + if randomize_change_idx and change_txout: + txouts.remove(change_txout) + txouts.insert(random.randrange(len(txouts) + 1), change_txout) + + tx = CMutableTransaction(txins, txouts) + + for idx, utxo in enumerate(utxos): + path = utxo['path'].replace("M/", "") + + key = self.primary_private_key.subkey_for_path(path) + redeemScript = CScript(x(utxo['redeem_script'])) + sighash = SignatureHash(redeemScript, tx, idx, SIGHASH_ALL) + + ckey = CBitcoinSecret(key.wif()) + + sig = ckey.sign(sighash) + struct.pack("B", SIGHASH_ALL) + + txins[idx].scriptSig = CScript([OP_0, sig, redeemScript]) + + signed = self.client.send_transaction(self.identifier, b2x(tx.serialize()), [utxo['path'] for utxo in utxos], check_fee=True) + + return signed['txid'] + + def do_discovery(self, gap=200): + balance_info = self.client.wallet_discovery(self.identifier, gap=gap) + + return balance_info['confirmed'], balance_info['unconfirmed'] + + def upgrade_key_index(self, key_index): + primary_public_key = self.primary_private_key.subkey_for_path("%d'.pub" % key_index) + + data = self.client.upgrade_key_index(self.identifier, key_index, (primary_public_key.hwif(), "M/%d'" % key_index)) + + self.key_index = key_index + for blocktrail_key_index, blocktrail_public_key in data['blocktrail_public_keys'].items(): + self.blocktrail_public_keys[str(blocktrail_key_index)] = BIP32Node.from_hwif(blocktrail_public_key[0]) + + def delete_wallet(self): + # can't right now because we can't create a signature + raise Exception("Not implemented") + + address, signature = self.create_checksum_signature() + result = self.client.delete_wallet(self.identifier, address, signature) + + return result and result['deleted'] + + def create_checksum_signature(self): + key = CBitcoinSecret(self.primary_private_key.wif()) + address = P2PKHBitcoinAddress.from_pubkey(key.pub) + + signature = None + + print(self.primary_private_key.wif(), str(address)) + + return str(address), signature + + def transactions(self, page=1, limit=20): + return self.client.wallet_transactions(self.identifier, page=page, limit=limit) + + def addresses(self, page=1, limit=20): + return self.client.wallet_addresses(self.identifier, page=page, limit=limit) + + def setup_webhook(self, url, identifier=None): + if identifier is None: + identifier = "WALLET-%s" % (self.identifier, ) + + return self.client.setup_wallet_webhook(self.identifier, identifier, url) + + def delete_webhook(self, identifier=None): + if identifier is None: + identifier = "WALLET-%s" % (self.identifier, ) + + return self.client.delete_wallet_webhook(self.identifier, identifier) diff --git a/examples/simple_payment_api_usage.py b/examples/simple_payment_api_usage.py new file mode 100644 index 0000000..689278d --- /dev/null +++ b/examples/simple_payment_api_usage.py @@ -0,0 +1,19 @@ +from __future__ import print_function +import blocktrail +from blocktrail.exceptions import ObjectNotFound + +client = blocktrail.APIClient("MY_APIKEY", "MY_APISECRET", testnet=True) + +try: + wallet = client.init_wallet("example-wallet", "example-strong-password") +except ObjectNotFound: + wallet, primary_mnemonic, backup_mnemonic, blocktrail_pubkeys = client.create_new_wallet("example-wallet", "example-strong-password", key_index=9999) + wallet.do_discovery() + +print(wallet.get_new_address_pair()) + +print(wallet.get_balance()) + +path, address = wallet.get_new_address_pair() + +print(wallet.pay([(address, blocktrail.to_satoshi(0.001))])) \ No newline at end of file diff --git a/setup.py b/setup.py index 074228e..7390ac8 100644 --- a/setup.py +++ b/setup.py @@ -7,10 +7,22 @@ sys.exit(1) setup( - name='blocktrail-sdk', - version='1.0.5', - description="BlockTrail's Developer Friendly API binding for Python", - long_description='This package allows interacting with the BlockTrail API', + name='blocktrail-sdk-beta', + version='1.0.6', + description="BlockTrail's Developer Friendly Bitcoin SDK", + long_description="""\ +BlockTrail's Developer Friendly Bitcoin SDK + + - simple bindings to the various data API endpoints + - block data and transactions + - transaction data + - address data and transactions + - latest price + - Contains Multi-Signature HD Wallet + +For examples and instructions on how to use, please see our official documentation at https://www.blocktrail.com/api/docs/lang/python +""", + keywords=["bitcoin", "sdk", "api", "payments", "crypto", "wallet", "multisig", "multisignature", "HD wallet"], maintainer='Ruben de Vries', maintainer_email='ruben@blocktrail.com', url='https://www.blocktrail.com/api/docs/lang/python', @@ -20,15 +32,18 @@ 'Environment :: Web Environment', 'Intended Audience :: Developers', 'Programming Language :: Python', + 'Programming Language :: Python :: 3', 'License :: OSI Approved :: MIT License' ], packages=["blocktrail"], install_requires=[ - 'httpsig >= 1.1.0', - 'pycrypto >= 2.6.1', - 'requests >= 2.4.3', - 'future >= 0.14.3', + 'httpsig-cffi == 15.0.0', + 'requests >= 2.4.3, < 2.5', + 'future >= 0.14.3, < 0.15', 'six >= 1.9.0', + 'pycoin == 0.52', + 'python-bitcoinlib == 0.2.1', + 'mnemonic == 0.12' ], test_suite="tests.get_tests", ) diff --git a/tests/cross_platform_test.py b/tests/cross_platform_test.py index 2c5c2b5..ee85a36 100644 --- a/tests/cross_platform_test.py +++ b/tests/cross_platform_test.py @@ -1,7 +1,14 @@ import unittest import hashlib -import httpsig.sign as sign -from httpsig.utils import parse_authorization_header + + +try: + import httpsig_cffi as sign + from httpsig_cffi.utils import parse_authorization_header +except: + import httpsig as sign + from httpsig.utils import parse_authorization_header + from requests.models import RequestEncodingMixin diff --git a/tests/wallet_test.py b/tests/wallet_test.py new file mode 100644 index 0000000..d091a32 --- /dev/null +++ b/tests/wallet_test.py @@ -0,0 +1,328 @@ +from __future__ import print_function + +import unittest +import os +import binascii +import time +import blocktrail +from bitcoin.core import x, b2x, lx +from blocktrail.exceptions import ObjectNotFound +from pycoin.key.BIP32Node import BIP32Node +from mnemonic import Mnemonic + + +class WalletTestCase(unittest.TestCase): + def setUp(self): + self.cleanup_data = {} + + def tearDown(self): + # cleanup any records that were created after each test + client = self.setup_api_client(debug=False) + + # webhooks + for webhook in self.cleanup_data.get('webhooks', []): + try: + client.delete_webhook(webhook) + except Exception: + pass + + # wallets + for wallet in self.cleanup_data.get('wallets', []): + try: + pass + # @TODO: delete_wallet is not implemented yet + # wallet.delete_wallet() + except Exception: + pass + + def setup_api_client(self, api_key=None, api_secret=None, debug=False): + if api_key is None: + api_key = os.environ.get('BLOCKTRAIL_SDK_APIKEY', 'EXAMPLE_BLOCKTRAIL_SDK_PYTHON_APIKEY') + if api_secret is None: + api_secret = os.environ.get('BLOCKTRAIL_SDK_APISECRET', 'EXAMPLE_BLOCKTRAIL_SDK_PYTHON_APISECRET') + + return blocktrail.APIClient(api_key, api_secret, testnet=True, debug=debug) + + def create_transaction_test_wallet(self, client, identifier="unittest-transaction"): + primary_mnemonic = "give pause forget seed dance crawl situate hole keen" + backup_mnemonic = "give pause forget seed dance crawl situate hole give" + passphrase = "password" + + return self.create_test_wallet(client, identifier, passphrase, primary_mnemonic, backup_mnemonic) + + def create_discovery_test_wallet(self, client, identifier="unittest-transaction", passphrase="password"): + primary_mnemonic = "give pause forget seed dance crawl situate hole kingdom" + backup_mnemonic = "give pause forget seed dance crawl situate hole course" + + return self.create_test_wallet(client, identifier, passphrase, primary_mnemonic, backup_mnemonic) + + def create_test_wallet(self, client, identifier, passphrase, primary_mnemonic, backup_mnemonic): + testnet = True + netcode = "XTN" if testnet else "BTC" + key_index = 9999 + primary_seed = Mnemonic.to_seed(primary_mnemonic, passphrase) + primary_private_key = BIP32Node.from_master_secret(primary_seed, netcode=netcode) + + primary_public_key = primary_private_key.subkey_for_path("%d'.pub" % key_index) + + backup_seed = Mnemonic.to_seed(backup_mnemonic, "") + backup_public_key = BIP32Node.from_master_secret(backup_seed, netcode=netcode).public_copy() + + checksum = blocktrail.APIClient.create_checksum(primary_private_key) + + result = client._create_new_wallet( + identifier=identifier, + primary_public_key=(primary_public_key.as_text(), "M/%d'" % key_index), + backup_public_key=(backup_public_key.as_text(), "M"), + primary_mnemonic=primary_mnemonic, + checksum=checksum, + key_index=key_index + ) + + blocktrail_public_keys = result['blocktrail_public_keys'] + key_index = result['key_index'] + + return blocktrail.Wallet( + client=client, + identifier=identifier, + primary_mnemonic=primary_mnemonic, + primary_private_key=primary_private_key, + backup_public_key=backup_public_key, + blocktrail_public_keys=blocktrail_public_keys, + key_index=key_index, + testnet=testnet + ) + + def get_random_test_identifier(self): + bytes = os.urandom(10) + time.time() + return "py-sdk-" + str(int(time.time())) + "-" + binascii.hexlify(bytes).decode("utf-8") + + def test_bip32(self): + master = "tpubD9q6vq9zdP3gbhpjs7n2TRvT7h4PeBhxg1Kv9jEc1XAss7429VenxvQTsJaZhzTk54gnsHRpgeeNMbm1QTag4Wf1QpQ3gy221GDuUCxgfeZ" + + key = BIP32Node.from_hwif(master) + + self.assertEqual(b2x(key.sec()), "022f6b9339309e89efb41ecabae60e1d40b7809596c68c03b05deb5a694e33cd26") + + self.assertEqual(key.subkey_for_path("0").hwif(), "tpubDAtJthHcm9MJwmHp4r2UwSTmiDYZWHbQUMqySJ1koGxQpRNSaJdyL2Ab8wwtMm5DsMMk3v68299LQE6KhT8XPQWzxPLK5TbTHKtnrmjV8Gg") + self.assertEqual(key.subkey_for_path("0/0").hwif(), "tpubDDfqpEKGqEVa5FbdLtwezc6Xgn81teTFFVA69ZfJBHp4UYmUmhqVZMmqXeJBDahvySZrPjpwMy4gKfNfrxuFHmzo1r6srB4MrsDKWbwEw3d") + + key = BIP32Node.from_master_secret(x("000102030405060708090a0b0c0d0e0f"), "XTN") + + self.assertEqual(key.subkey_for_path("0'/1/2'/2/1000000000").hwif(), "tpubDHNy3kAG39ThyiwwsgoKY4iRenXDRtce8qdCFJZXPMCJg5dsCUHayp84raLTpvyiNA9sXPob5rgqkKvkN8S7MMyXbnEhGJMW64Cf4vFAoaF") + + def test_create_wallet(self): + client = self.setup_api_client() + + identifier = self.get_random_test_identifier() + + wallet = self.create_transaction_test_wallet(client, identifier) + + self.cleanup_data.setdefault('wallets', []).append(wallet) + + wallets = client.all_wallets() + self.assertTrue(len(wallets) > 0) + + self.assertEqual(wallet.primary_mnemonic, "give pause forget seed dance crawl situate hole keen") + self.assertEqual(wallet.identifier, identifier) + self.assertEqual(wallet.blocktrail_public_keys['9999'].as_text(), "tpubD9q6vq9zdP3gbhpjs7n2TRvT7h4PeBhxg1Kv9jEc1XAss7429VenxvQTsJaZhzTk54gnsHRpgeeNMbm1QTag4Wf1QpQ3gy221GDuUCxgfeZ") + + self.assertEqual(wallet.get_address_by_path("M/9999'/0/0"), "2MzyKviSL6pnWxkbHV7ecFRE3hWKfzmT8WS") + + path, address = wallet.get_new_address_pair() + self.assertEqual(path, "M/9999'/0/0") + self.assertEqual(address, "2MzyKviSL6pnWxkbHV7ecFRE3hWKfzmT8WS") + + path, address = wallet.get_new_address_pair() + self.assertEqual(path, "M/9999'/0/1") + self.assertEqual(address, "2N65RcfKHiKQcPGZAA2QVeqitJvAQ8HroHD") + + self.assertEqual(wallet.get_address_by_path("M/9999'/0/1"), "2N65RcfKHiKQcPGZAA2QVeqitJvAQ8HroHD") + self.assertEqual(wallet.get_address_by_path("M/9999'/0/6"), "2MynrezSyqCq1x5dMPtRDupTPA4sfVrNBKq") + self.assertEqual(wallet.get_address_by_path("M/9999'/0/44"), "2N5eqrZE7LcfRyCWqpeh1T1YpMdgrq8HWzh") + + def test_wallet_transaction(self): + client = self.setup_api_client() + + wallet = client.init_wallet("unittest-transaction", "password") + + confirmed, unconfirmed = wallet.get_balance() + self.assertGreater(unconfirmed + confirmed, 0) + self.assertGreater(confirmed, 0) + + path, address = wallet.get_new_address_pair() + self.assertTrue("M/9999'/0" in path) + # self.assertEqual(address, "2MzyKviSL6pnWxkbHV7ecFRE3hWKfzmT8WS" # validate address) + + value = blocktrail.to_satoshi(0.0002) + txhash = wallet.pay([(address, value)]) + + self.assertTrue(txhash) + + time.sleep(1) + + tx = client.transaction(txhash) + + self.assertTrue(tx) + self.assertEqual(tx['hash'], txhash) + self.assertTrue(len(tx['outputs']) <= 2) + self.assertTrue(value in map(lambda o: o['value'], tx['outputs'])) + + def test_discovery_and_key_index_upgrade(self): + client = self.setup_api_client() + + identifier = self.get_random_test_identifier() + + wallet = self.create_discovery_test_wallet(client, identifier) + + self.cleanup_data.setdefault('wallets', []).append(wallet) + + self.assertEqual(wallet.primary_mnemonic, "give pause forget seed dance crawl situate hole kingdom") + self.assertEqual(wallet.identifier, identifier) + self.assertEqual(wallet.blocktrail_public_keys['9999'].as_text(), "tpubD9q6vq9zdP3gbhpjs7n2TRvT7h4PeBhxg1Kv9jEc1XAss7429VenxvQTsJaZhzTk54gnsHRpgeeNMbm1QTag4Wf1QpQ3gy221GDuUCxgfeZ") + + self.assertEqual(wallet.get_address_by_path("M/9999'/0/0"), "2Mtfn5S9tVWnnHsBQixCLTsCAPFHvfhu6bM") + + path, address = wallet.get_new_address_pair() + self.assertEqual(path, "M/9999'/0/0") + self.assertEqual(address, "2Mtfn5S9tVWnnHsBQixCLTsCAPFHvfhu6bM") + + path, address = wallet.get_new_address_pair() + self.assertEqual(path, "M/9999'/0/1") + self.assertEqual(address, "2NG49GDkm5qCYvDFi4cxAnkSho8qLbEz6C4") + + confirmed, unconfirmed = wallet.do_discovery(gap=50) + self.assertGreater(confirmed + unconfirmed, 0) + + wallet.upgrade_key_index(10000) + + self.assertEqual(wallet.blocktrail_public_keys['10000'].as_text(), "tpubD9m9hziKhYQExWgzMUNXdYMNUtourv96sjTUS9jJKdo3EDJAnCBJooMPm6vGSmkNTNAmVt988dzNfNY12YYzk9E6PkA7JbxYeZBFy4XAaCp") + + self.assertEqual(wallet.get_address_by_path("M/10000'/0/0"), "2N9ZLKXgs12JQKXvLkngn7u9tsYaQ5kXJmk") + + path, address = wallet.get_new_address_pair() + self.assertEqual(path, "M/10000'/0/0") + self.assertEqual(address, "2N9ZLKXgs12JQKXvLkngn7u9tsYaQ5kXJmk") + + def test_list_wallet_txs_addrs(self): + client = self.setup_api_client() + + wallet = client.init_wallet("unittest-transaction", "password") + + txs = wallet.transactions(page=1, limit=23) + + self.assertEqual(len(txs['data']), 23) + self.assertEqual(txs['data'][0]['hash'], '2cb21783635a5f22e9934b8c3262146b42d251dfb14ee961d120936a6c40fe89') + + addresses = wallet.addresses(page=1, limit=23) + + self.assertEqual(len(addresses['data']), 23) + self.assertEqual(addresses['data'][0]['address'], '2MzyKviSL6pnWxkbHV7ecFRE3hWKfzmT8WS') + + def test_bad_password_wallet(self): + client = self.setup_api_client() + + identifier = self.get_random_test_identifier() + + wallet = self.create_discovery_test_wallet(client, identifier, "badpassword") + + self.cleanup_data.setdefault('wallets', []).append(wallet) + + self.assertEqual(wallet.primary_mnemonic, "give pause forget seed dance crawl situate hole kingdom") + self.assertEqual(wallet.identifier, identifier) + self.assertEqual(wallet.blocktrail_public_keys['9999'].as_text(), "tpubD9q6vq9zdP3gbhpjs7n2TRvT7h4PeBhxg1Kv9jEc1XAss7429VenxvQTsJaZhzTk54gnsHRpgeeNMbm1QTag4Wf1QpQ3gy221GDuUCxgfeZ") + + self.assertEqual(wallet.get_address_by_path("M/9999'/0/0"), "2N9SGrV4NKRjdACYvHLPpy2oiPrxTPd44rg") + + path, address = wallet.get_new_address_pair() + self.assertEqual(path, "M/9999'/0/0") + self.assertEqual(address, "2N9SGrV4NKRjdACYvHLPpy2oiPrxTPd44rg") + + path, address = wallet.get_new_address_pair() + self.assertEqual(path, "M/9999'/0/1") + self.assertEqual(address, "2NDq3DRy9E3YgHDA3haPJj3FtUS6V93avkf") + + confirmed, unconfirmed = wallet.do_discovery(gap=50) + self.assertEqual(confirmed + unconfirmed, 0) + + def test_new_blank_wallet(self): + client = self.setup_api_client() + + identifier = self.get_random_test_identifier() + + # wallet shouldn't exist yet + wallet = None + try: + wallet = client.init_wallet(identifier, 'password') + except ObjectNotFound as e: + pass + self.assertFalse(wallet) + + # create wallet + wallet, primary_mnemonic, backup_mnemonic, blocktrail_pubkeys = client.create_new_wallet(identifier, "password", 9999) + + self.cleanup_data.setdefault('wallets', []).append(wallet) + + # wallet shouldn't be able to pay + txhash = None + try: + txhash = wallet.pay({"2N6Fg6T74Fcv1JQ8FkPJMs8mYmbm9kitTxy": blocktrail.to_satoshi(0.001)}) + except: + pass + self.assertFalse(txhash) + + # same wallet with bad password shouldn't be initialized + wallet = None + try: + wallet = client.init_wallet(identifier, "badpassword") + except: + pass + self.assertFalse(wallet) + + def test_webhook_for_wallet(self): + client = self.setup_api_client() + + identifier = self.get_random_test_identifier() + + # create wallet + wallet, primary_mnemonic, backup_mnemonic, blocktrail_pubkeys = client.create_new_wallet(identifier, "password", 9999) + + self.cleanup_data.setdefault('wallets', []).append(wallet) + + confirmed, unconfirmed = wallet.get_balance() + + self.assertEqual(confirmed + unconfirmed, 0) + + webhook = wallet.setup_webhook("https://www.blocktrail.com/webhook-test") + self.assertEqual(webhook['url'], "https://www.blocktrail.com/webhook-test") + self.assertEqual(webhook['identifier'], "WALLET-%s" % (wallet.identifier, )) + + self.assertTrue(wallet.delete_webhook()) + + webhook_identifier = self.get_random_test_identifier() + webhook = wallet.setup_webhook("https://www.blocktrail.com/webhook-test", identifier=webhook_identifier) + self.assertEqual(webhook['url'], "https://www.blocktrail.com/webhook-test") + self.assertEqual(webhook['identifier'], webhook_identifier) + + path, address = wallet.get_new_address_pair() + + events = client.webhook_events(webhook_identifier) + + self.assertIn(address, [event['address'] for event in events['data']]) + + # @TODO: delete_wallet is not implemented yet + # self.assertTrue(wallet.delete_wallet()) + + # webhook should already be deleted so we shouldn't be able to delete it again + # deleted = None + # try: + # deleted = wallet.delete_webhook(webhook_identifier) + # except: + # pass + + # self.assertFalse(deleted) + + +if __name__ == "__main__": + unittest.main()