19944567212

Committed 04 Dec 2025 09:29PM UTC coverage: 90.921% (-0.02%) from 90.944%

Build # 19944567212

Build Type

push

github

Committed by Cryp Toon

Commit Message

RELEASE 0.7.6 - MESSAGE SIGNING AND DETERMINISTIC SIGNATURES

Coverage Stats

8042 of 8845 relevant lines covered (90.92%)

1.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.48

/bitcoinlib/services/bitgo.py

# -*- coding: utf-8 -*-
#
#    BitcoinLib - Python Cryptocurrency Library
#    BitGo Client
#    © 2017-2019 July - 1200 Web Development <http://1200wd.com/>
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as
#    published by the Free Software Foundation, either version 3 of the
#    License, or (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import logging
from datetime import datetime, timezone
from bitcoinlib.main import MAX_TRANSACTIONS
from bitcoinlib.services.baseclient import BaseClient, ClientError
from bitcoinlib.transactions import Transaction

_logger = logging.getLogger(__name__)

PROVIDERNAME = 'bitgo'
LIMIT_TX = 49


class BitGoClient(BaseClient):

    def __init__(self, network, base_url, denominator, *args):
        super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)

    def compose_request(self, category, data, cmd='', variables=None, method='get'):
        if data:
            data = '/' + data
        url_path = category + data
        if cmd != '':
            url_path += '/' + cmd
        return self.request(url_path, variables, method=method)

    # def getbalance(self, addresslist):
    #     balance = 0
    #     for address in addresslist:
    #         res = self.compose_request('address', address)
    #         balance += res['balance']
    #     return balance

    def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        utxos = []
        skip = 0
        total = 1
        while total > skip:
            variables = {'limit': 100, 'skip': skip}
            res = self.compose_request('address', address, 'unspents', variables)
            for utxo in res['unspents'][::-1]:
                if utxo['tx_hash'] == after_txid:
                    break
                utxos.append(
                    {
                        'address': utxo['address'],
                        'txid': utxo['tx_hash'],
                        'confirmations': utxo['confirmations'],
                        'output_n': utxo['tx_output_n'],
                        'input_n': 0,
                        'block_height': round(int(utxo['blockHeight'])) if utxo['blockHeight'] else None,
                        'fee': None,
                        'size': 0,
                        'value': round(utxo['value'] * self.units),
                        'script': utxo['script'],
                        'date': datetime.strptime(utxo['date'], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc)
                     }
                )
            total = res['total']
            skip = res['start'] + res['count']
            if skip > 2000:
                _logger.info("BitGoClient: UTXO's list has been truncated, list is incomplete")
                break
        return utxos[::-1][:limit]

    # RAW TRANSACTION DOES NOT CONTAIN CORRECT RAW TRANSACTION (MISSING SIGS)
    # def gettransaction(self, txid):
    #     tx = self.compose_request('tx', txid)
    #     t = Transaction.parse_hex(tx['hex'], strict=self.strict, network=self.network)
    #     t.status = 'unconfirmed'
    #     t.date = None
    #     if tx['confirmations']:
    #         t.status = 'confirmed'
    #         t.date = datetime.strptime(tx['date'], "%Y-%m-%dT%H:%M:%S.%fZ")
    #     t.confirmations = tx['confirmations']
    #     if 'height' in tx:
    #         t.block_height = tx['height']
    #         t.block_hash = tx['blockhash']
    #     t.fee = tx['fee'] or 0
    #     t.rawtx = bytes.fromhex(tx['hex'])
    #     t.size = len(tx['hex']) // 2
    #     t.network = self.network
    #     if t.coinbase:
    #         input_values = []
    #         t.input_total = t.output_total
    #     else:
    #         input_values = [(inp['account'], -inp['value']) for inp in tx['entries'] if inp['value'] < 0]
    #         if len(input_values) >= 49:
    #             raise ClientError("More then 49 transaction inputs not supported by bitgo")
    #         t.input_total = sum([x[1] for x in input_values])
    #     for i in t.inputs:
    #         if not i.address and not t.coinbase:
    #             raise ClientError("Address missing in input. Provider might not support segwit transactions")
    #         if len(t.inputs) != len(input_values):
    #             i.value = 0
    #             continue
    #         value = [x[1] for x in input_values if x[0] == i.address]
    #         if len(value) != 1:
    #             _logger.info("BitGoClient: Address %s input value should be found exactly 1 times in value list" %
    #                             i.address)
    #             i.value = None
    #         else:
    #             i.value = value[0]
    #     for o in t.outputs:
    #         o.spent = None
    #     if t.input_total != t.output_total + t.fee:
    #         t.input_total = t.output_total + t.fee
    #     return t

    # RAW TRANSACTION DOES NOT CONTAIN CORRECT RAW TRANSACTION (MISSING SIGS)
    # def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):
    #     txs = []
    #     txids = []
    #     skip = 0
    #     total = 1
    #     while total > skip:
    #         variables = {'limit': LIMIT_TX, 'skip': skip}
    #         res = self.compose_request('address', address, 'tx', variables)
    #         for tx in res['transactions']:
    #             if tx['id'] not in txids:
    #                 txids.insert(0, tx['id'])
    #         total = res['total']
    #         # if total > 2000:
    #         #     raise ClientError("BitGoClient: Transactions list limit exceeded > 2000")
    #         skip = res['start'] + res['count']
    #         if len(txids) > limit:
    #             break
    #     if after_txid:
    #         txids = txids[txids.index(after_txid) + 1:]
    #     for txid in txids[:limit]:
    #         txs.append(self.gettransaction(txid))
    #     return txs

    # RAW TRANSACTION DOES NOT CONTAIN CORRECT RAW TRANSACTION (MISSING SIGS)
    # def getrawtransaction(self, txid):
    #     tx = self.compose_request('tx', txid)
    #     t = Transaction.parse_hex(tx['hex'], strict=self.strict, network=self.network)
    #     for i in t.inputs:
    #         if not i.address:
    #             raise ClientError("Address missing in input. Provider might not support segwit transactions")
    #     return tx['hex']

    # def sendrawtransaction

    def estimatefee(self, blocks):
        res = self.compose_request('tx', 'fee', variables={'numBlocks': blocks})
        return res['feePerKb']

    def blockcount(self):
        if self.network == 'testnet':
            raise ClientError('Providers return incorrect blockcount for testnet')
        return self.compose_request('block', 'latest')['height']

    # def mempool

    # def getblock(self, blockid, parse_transactions, page, limit):
    #     bd = self.compose_request('block', str(blockid))
    #     if parse_transactions:
    #         txs = []
    #         for txid in bd['transactions'][(page-1)*limit:page*limit]:
    #             try:
    #                 txs.append(self.gettransaction(txid))
    #             except Exception as e:
    #                 _logger.error("Could not parse tx %s with error %s" % (txid, e))
    #     else:
    #         txs = bd['transactions']
    #
    #     block = {
    #         'bits': None,
    #         'depth': None,
    #         'hash': bd['id'],
    #         'height': bd['height'],
    #         'merkle_root': bd['merkleRoot'],
    #         'nonce': bd['nonce'],
    #         'prev_block': bd['previous'],
    #         'time': datetime.strptime(bd['date'], "%Y-%m-%dT%H:%M:%S.%fZ").replace(microsecond=0),
    #         'total_txs': len(bd['transactions']),
    #         'txs': txs,
    #         'version': bd['version'],
    #         'page': page,
    #         'pages': None if not limit else int(len(bd['transactions']) // limit) + (len(bd['transactions']) % limit > 0),
    #         'limit': limit
    #     }
    #     return block

    # def isspent(self, txid, index):

    # def getinfo(self):

1	# -- coding: utf-8 --
2	#
3	# BitcoinLib - Python Cryptocurrency Library
4	# BitGo Client
5	# © 2017-2019 July - 1200 Web Development <http://1200wd.com/>
6	#
7	# This program is free software: you can redistribute it and/or modify
8	# it under the terms of the GNU Affero General Public License as
9	# published by the Free Software Foundation, either version 3 of the
10	# License, or (at your option) any later version.
11	#
12	# This program is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	# GNU Affero General Public License for more details.
16	#
17	# You should have received a copy of the GNU Affero General Public License
18	# along with this program. If not, see <http://www.gnu.org/licenses/>.
19	#
20
21	import logging	2✔
22	from datetime import datetime, timezone	2✔
23	from bitcoinlib.main import MAX_TRANSACTIONS	2✔
24	from bitcoinlib.services.baseclient import BaseClient, ClientError	2✔
25	from bitcoinlib.transactions import Transaction	2✔
26
27	_logger = logging.getLogger(__name__)	2✔
28
29	PROVIDERNAME = 'bitgo'	2✔
30	LIMIT_TX = 49	2✔
31
32
33	class BitGoClient(BaseClient):	2✔
34
35	def __init__(self, network, base_url, denominator, *args):	2✔
36	super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)	2✔
37
38	def compose_request(self, category, data, cmd='', variables=None, method='get'):	2✔
39	if data:	2✔
40	data = '/' + data	2✔
41	url_path = category + data	2✔
42	if cmd != '':	2✔
43	url_path += '/' + cmd	1✔
44	return self.request(url_path, variables, method=method)	2✔
45
46	# def getbalance(self, addresslist):
47	# balance = 0
48	# for address in addresslist:
49	# res = self.compose_request('address', address)
50	# balance += res['balance']
51	# return balance
52
53	def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
54	utxos = []	1✔
55	skip = 0	1✔
56	total = 1	1✔
57	while total > skip:	1✔
58	variables = {'limit': 100, 'skip': skip}	1✔
59	res = self.compose_request('address', address, 'unspents', variables)	1✔
60	for utxo in res['unspents'][::-1]:	1✔
61	if utxo['tx_hash'] == after_txid:	1✔
62	break	×
63	utxos.append(	1✔
64	{
65	'address': utxo['address'],
66	'txid': utxo['tx_hash'],
67	'confirmations': utxo['confirmations'],
68	'output_n': utxo['tx_output_n'],
69	'input_n': 0,
70	'block_height': round(int(utxo['blockHeight'])) if utxo['blockHeight'] else None,
71	'fee': None,
72	'size': 0,
73	'value': round(utxo['value'] * self.units),
74	'script': utxo['script'],
75	'date': datetime.strptime(utxo['date'], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc)
76	}
77	)
78	total = res['total']	1✔
79	skip = res['start'] + res['count']	1✔
80	if skip > 2000:	1✔
81	_logger.info("BitGoClient: UTXO's list has been truncated, list is incomplete")	×
82	break	×
83	return utxos[::-1][:limit]	1✔
84
85	# RAW TRANSACTION DOES NOT CONTAIN CORRECT RAW TRANSACTION (MISSING SIGS)
86	# def gettransaction(self, txid):
87	# tx = self.compose_request('tx', txid)
88	# t = Transaction.parse_hex(tx['hex'], strict=self.strict, network=self.network)
89	# t.status = 'unconfirmed'
90	# t.date = None
91	# if tx['confirmations']:
92	# t.status = 'confirmed'
93	# t.date = datetime.strptime(tx['date'], "%Y-%m-%dT%H:%M:%S.%fZ")
94	# t.confirmations = tx['confirmations']
95	# if 'height' in tx:
96	# t.block_height = tx['height']
97	# t.block_hash = tx['blockhash']
98	# t.fee = tx['fee'] or 0
99	# t.rawtx = bytes.fromhex(tx['hex'])
100	# t.size = len(tx['hex']) // 2
101	# t.network = self.network
102	# if t.coinbase:
103	# input_values = []
104	# t.input_total = t.output_total
105	# else:
106	# input_values = [(inp['account'], -inp['value']) for inp in tx['entries'] if inp['value'] < 0]
107	# if len(input_values) >= 49:
108	# raise ClientError("More then 49 transaction inputs not supported by bitgo")
109	# t.input_total = sum([x[1] for x in input_values])
110	# for i in t.inputs:
111	# if not i.address and not t.coinbase:
112	# raise ClientError("Address missing in input. Provider might not support segwit transactions")
113	# if len(t.inputs) != len(input_values):
114	# i.value = 0
115	# continue
116	# value = [x[1] for x in input_values if x[0] == i.address]
117	# if len(value) != 1:
118	# _logger.info("BitGoClient: Address %s input value should be found exactly 1 times in value list" %
119	# i.address)
120	# i.value = None
121	# else:
122	# i.value = value[0]
123	# for o in t.outputs:
124	# o.spent = None
125	# if t.input_total != t.output_total + t.fee:
126	# t.input_total = t.output_total + t.fee
127	# return t
128
129	# RAW TRANSACTION DOES NOT CONTAIN CORRECT RAW TRANSACTION (MISSING SIGS)
130	# def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):
131	# txs = []
132	# txids = []
133	# skip = 0
134	# total = 1
135	# while total > skip:
136	# variables = {'limit': LIMIT_TX, 'skip': skip}
137	# res = self.compose_request('address', address, 'tx', variables)
138	# for tx in res['transactions']:
139	# if tx['id'] not in txids:
140	# txids.insert(0, tx['id'])
141	# total = res['total']
142	# # if total > 2000:
143	# # raise ClientError("BitGoClient: Transactions list limit exceeded > 2000")
144	# skip = res['start'] + res['count']
145	# if len(txids) > limit:
146	# break
147	# if after_txid:
148	# txids = txids[txids.index(after_txid) + 1:]
149	# for txid in txids[:limit]:
150	# txs.append(self.gettransaction(txid))
151	# return txs
152
153	# RAW TRANSACTION DOES NOT CONTAIN CORRECT RAW TRANSACTION (MISSING SIGS)
154	# def getrawtransaction(self, txid):
155	# tx = self.compose_request('tx', txid)
156	# t = Transaction.parse_hex(tx['hex'], strict=self.strict, network=self.network)
157	# for i in t.inputs:
158	# if not i.address:
159	# raise ClientError("Address missing in input. Provider might not support segwit transactions")
160	# return tx['hex']
161
162	# def sendrawtransaction
163
164	def estimatefee(self, blocks):	2✔
165	res = self.compose_request('tx', 'fee', variables={'numBlocks': blocks})	1✔
166	return res['feePerKb']	1✔
167
168	def blockcount(self):	2✔
169	if self.network == 'testnet':	1✔
170	raise ClientError('Providers return incorrect blockcount for testnet')	×
171	return self.compose_request('block', 'latest')['height']	1✔
172
173	# def mempool
174
175	# def getblock(self, blockid, parse_transactions, page, limit):
176	# bd = self.compose_request('block', str(blockid))
177	# if parse_transactions:
178	# txs = []
179	# for txid in bd['transactions'][(page-1)limit:pagelimit]:
180	# try:
181	# txs.append(self.gettransaction(txid))
182	# except Exception as e:
183	# _logger.error("Could not parse tx %s with error %s" % (txid, e))
184	# else:
185	# txs = bd['transactions']
186	#
187	# block = {
188	# 'bits': None,
189	# 'depth': None,
190	# 'hash': bd['id'],
191	# 'height': bd['height'],
192	# 'merkle_root': bd['merkleRoot'],
193	# 'nonce': bd['nonce'],
194	# 'prev_block': bd['previous'],
195	# 'time': datetime.strptime(bd['date'], "%Y-%m-%dT%H:%M:%S.%fZ").replace(microsecond=0),
196	# 'total_txs': len(bd['transactions']),
197	# 'txs': txs,
198	# 'version': bd['version'],
199	# 'page': page,
200	# 'pages': None if not limit else int(len(bd['transactions']) // limit) + (len(bd['transactions']) % limit > 0),
201	# 'limit': limit
202	# }
203	# return block
204
205	# def isspent(self, txid, index):
206
207	# def getinfo(self):

1200wd / bitcoinlib / 19944567212

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous