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

Run Details

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

91.82

/bitcoinlib/services/blockstream.py

# -*- coding: utf-8 -*-
#
#    BitcoinLib - Python Cryptocurrency Library
#    BlockstreamClient client
#    © 2019 November - 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
from bitcoinlib.transactions import Transaction


PROVIDERNAME = 'blockstream'
# Please note: In the Blockstream API, the first couple of Bitcoin blocks are not correctly indexed,
# so transactions from these blocks are missing.

_logger = logging.getLogger(__name__)


class BlockstreamClient(BaseClient):

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

    def compose_request(self, function, data='', parameter='', parameter2='', variables=None, post_data='', method='get'):
        url_path = function
        if data:
            url_path += '/' + data
        if parameter:
            url_path += '/' + parameter
        if parameter2:
            url_path += '/' + parameter2
        if variables is None:
            variables = {}
        if self.api_key:
            variables.update({'token': self.api_key})
        return self.request(url_path, variables, method, post_data=post_data)

    def getbalance(self, addresslist):
        balance = 0
        for address in addresslist:
            res = self.compose_request('address', data=address)
            balance += (res['chain_stats']['funded_txo_sum'] - res['chain_stats']['spent_txo_sum'])
        return balance

    def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        res = self.compose_request('address', address, 'utxo')
        self.latest_block = self.blockcount() if not self.latest_block else self.latest_block
        utxos = []
        res = sorted(res, key=lambda k: 0 if 'block_height' not in k['status'] else k['status']['block_height'])
        for a in res:
            confirmations = 0
            block_height = None
            if 'block_height' in a['status']:
                block_height = a['status']['block_height']
                confirmations = self.latest_block - block_height
            utxos.append({
                'address': address,
                'txid': a['txid'],
                'confirmations': confirmations,
                'output_n': a['vout'],
                'input_n': 0,
                'block_height': block_height,
                'fee': None,
                'size': 0,
                'value': a['value'],
                'script': '',
                'date': None if 'block_time' not in a['status'] else
                          datetime.fromtimestamp(a['status']['block_time'], timezone.utc)
            })
            if a['txid'] == after_txid:
                utxos = []
        return utxos[:limit]

    def _parse_transaction(self, tx):
        confirmations = 0
        block_height = None
        if 'block_height' in tx['status']:
            block_height = tx['status']['block_height']
            confirmations = self.latest_block - block_height
        status = 'unconfirmed'
        if tx['status']['confirmed']:
            status = 'confirmed'
        fee = None if 'fee' not in tx else tx['fee']
        witness_type = 'legacy'
        if tx['size'] * 4 > tx['weight']:
            witness_type = 'segwit'

        t = Transaction(locktime=tx['locktime'], version=tx['version'], network=self.network,
                        fee=fee, size=tx['size'], txid=tx['txid'],
                        date=None if 'block_time' not in tx['status'] else
                        datetime.fromtimestamp(tx['status']['block_time'], timezone.utc),
                        confirmations=confirmations, block_height=block_height, status=status,
                        coinbase=tx['vin'][0]['is_coinbase'], witness_type=witness_type)
        index_n = 0
        for ti in tx['vin']:
            if tx['vin'][0]['is_coinbase']:
                t.add_input(prev_txid=ti['txid'], output_n=ti['vout'], index_n=index_n, witness_type=witness_type,
                            unlocking_script=ti['scriptsig'], value=0, sequence=ti['sequence'], strict=self.strict)
            else:
                witnesses = []
                if 'witness' in ti:
                    witnesses = [bytes.fromhex(w) for w in ti['witness']]
                t.add_input(prev_txid=ti['txid'], output_n=ti['vout'], index_n=index_n,
                            unlocking_script=ti['scriptsig'], value=ti['prevout']['value'],
                            address='' if 'scriptpubkey_address' not in ti['prevout']
                            else ti['prevout']['scriptpubkey_address'], sequence=ti['sequence'],
                            locking_script=ti['prevout']['scriptpubkey'], witnesses=witnesses, strict=self.strict)
            index_n += 1
        index_n = 0
        if len(tx['vout']) > 101:
            # Every output needs an extra query, stop execution if there are too many transaction outputs
            return False
        for to in tx['vout']:
            address = ''
            if 'scriptpubkey_address' in to:
                address = to['scriptpubkey_address']
            spent = self.isspent(t.txid, index_n)
            t.add_output(value=to['value'], address=address, lock_script=to['scriptpubkey'],
                         output_n=index_n, spent=spent, strict=self.strict)
            index_n += 1
        if 'segwit' in [i.witness_type for i in t.inputs] or 'p2sh-segwit' in [i.witness_type for i in t.inputs]:
            t.witness_type = 'segwit'
        t.update_totals()
        t.size = tx['size']
        return t

    def gettransaction(self, txid):
        tx = self.compose_request('tx', txid)
        return self._parse_transaction(tx)

    def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        self.latest_block = self.blockcount() if not self.latest_block else self.latest_block
        prtxs = []
        before_txid = ''
        while True:
            parameter = 'txs'
            if before_txid:
                parameter = 'txs/chain/%s' % before_txid
            res = self.compose_request('address', address, parameter)
            prtxs += res
            if len(res) == 25:
                before_txid = res[-1:][0]['txid']
            else:
                break
            if len(prtxs) > limit:
                break
        txs = []
        if len(set([x['status'].get('block_height', '-1') for x in prtxs])) > 1:
            prtxs.sort(key=lambda x: x['status'].get('block_height', '-1'))
        else:
            prtxs =  prtxs[::-1]
        for tx in prtxs:
            t = self._parse_transaction(tx)
            if t:
                txs.append(t)
            if t.txid == after_txid:
                txs = []
            if len(txs) > limit:
                break
        return txs[:limit]

    def getrawtransaction(self, txid):
        return self.compose_request('tx', txid, 'hex')

    def sendrawtransaction(self, rawtx):
        res = self.compose_request('tx', post_data=rawtx, method='post')
        return {
            'txid': res,
            'response_dict': res
        }

    def estimatefee(self, blocks):
        est = self.compose_request('fee-estimates')
        closest = (sorted([int(i) - blocks for i in est.keys() if int(i) - blocks >= 0]))
        # FIXME: temporary fix for too low testnet tx fees:
        if self.network.name == 'testnet':
            return 2000
        if closest:
            return round(est[str(closest[0] + blocks)] * 1000)
        else:
            return round(est[str(sorted([int(i) for i in est.keys()])[-1:][0])] * 1000)

    def blockcount(self):
        return self.compose_request('blocks', 'tip', 'height')

    def mempool(self, txid):
        if txid:
            t = self.gettransaction(txid)
            if t and not t.confirmations:
                return [t.txid]
            else:
                return []
        else:
            return self.compose_request('mempool', 'txids')

    def getblock(self, blockid, parse_transactions, page, limit):
        if isinstance(blockid, int):
            blockid = self.compose_request('block-height', str(blockid))
        if (page == 1 and limit == 10) or limit > 25:
            limit = 25
        # elif page > 1:
        #     if limit % 25 != 0:
        #         return False
        bd = self.compose_request('block', blockid)
        btxs = self.compose_request('block', blockid, 'txs', str((page-1)*limit))
        if parse_transactions:
            txs = []
            self.latest_block = self.blockcount() if not self.latest_block else self.latest_block
            for tx in btxs[:limit]:
                # try:
                txs.append(self._parse_transaction(tx))
                # except Exception as e:
                #     _logger.error("Could not parse tx %s with error %s" % (tx['txid'], e))
        else:
            txs = [tx['txid'] for tx in btxs]

        block = {
            'bits': bd['bits'],
            'depth': None,
            'block_hash': bd['id'],
            'height': bd['height'],
            'merkle_root': bd['merkle_root'],
            'nonce': bd['nonce'],
            'prev_block': bd['previousblockhash'],
            'time': bd['timestamp'],
            'tx_count': bd['tx_count'],
            'txs': txs,
            'version': bd['version'],
            'page': page,
            'pages': None if not limit else int(bd['tx_count'] // limit) + (bd['tx_count'] % limit > 0),
            'limit': limit
        }
        return block

    def getrawblock(self, blockid):
        if isinstance(blockid, int):
            blockid = self.compose_request('block-height', str(blockid))
        rawblock = self.compose_request('block', blockid, 'raw')
        hexrawblock = rawblock.hex()
        return hexrawblock

    def isspent(self, txid, output_n):
        res = self.compose_request('tx', txid, 'outspend', str(output_n))
        return 1 if res['spent'] else 0

    # def getinfo(self):

1	# -- coding: utf-8 --
2	#
3	# BitcoinLib - Python Cryptocurrency Library
4	# BlockstreamClient client
5	# © 2019 November - 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	2✔
25	from bitcoinlib.transactions import Transaction	2✔
26
27
28	PROVIDERNAME = 'blockstream'	2✔
29	# Please note: In the Blockstream API, the first couple of Bitcoin blocks are not correctly indexed,
30	# so transactions from these blocks are missing.
31
32	_logger = logging.getLogger(__name__)	2✔
33
34
35	class BlockstreamClient(BaseClient):	2✔
36
37	def __init__(self, network, base_url, denominator, *args):	2✔
38	super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)	2✔
39
40	def compose_request(self, function, data='', parameter='', parameter2='', variables=None, post_data='', method='get'):	2✔
41	url_path = function	2✔
42	if data:	2✔
43	url_path += '/' + data	2✔
44	if parameter:	2✔
45	url_path += '/' + parameter	2✔
46	if parameter2:	2✔
47	url_path += '/' + parameter2	2✔
48	if variables is None:	2✔
49	variables = {}	2✔
50	if self.api_key:	2✔
51	variables.update({'token': self.api_key})	×
52	return self.request(url_path, variables, method, post_data=post_data)	2✔
53
54	def getbalance(self, addresslist):	2✔
55	balance = 0	2✔
56	for address in addresslist:	2✔
57	res = self.compose_request('address', data=address)	2✔
58	balance += (res['chain_stats']['funded_txo_sum'] - res['chain_stats']['spent_txo_sum'])	2✔
59	return balance	2✔
60
61	def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
62	res = self.compose_request('address', address, 'utxo')	2✔
63	self.latest_block = self.blockcount() if not self.latest_block else self.latest_block	2✔
64	utxos = []	2✔
65	res = sorted(res, key=lambda k: 0 if 'block_height' not in k['status'] else k['status']['block_height'])	2✔
66	for a in res:	2✔
67	confirmations = 0	2✔
68	block_height = None	2✔
69	if 'block_height' in a['status']:	2✔
70	block_height = a['status']['block_height']	2✔
71	confirmations = self.latest_block - block_height	2✔
72	utxos.append({	2✔
73	'address': address,
74	'txid': a['txid'],
75	'confirmations': confirmations,
76	'output_n': a['vout'],
77	'input_n': 0,
78	'block_height': block_height,
79	'fee': None,
80	'size': 0,
81	'value': a['value'],
82	'script': '',
83	'date': None if 'block_time' not in a['status'] else
84	datetime.fromtimestamp(a['status']['block_time'], timezone.utc)
85	})
86	if a['txid'] == after_txid:	2✔
87	utxos = []	2✔
88	return utxos[:limit]	2✔
89
90	def _parse_transaction(self, tx):	2✔
91	confirmations = 0	2✔
92	block_height = None	2✔
93	if 'block_height' in tx['status']:	2✔
94	block_height = tx['status']['block_height']	2✔
95	confirmations = self.latest_block - block_height	2✔
96	status = 'unconfirmed'	2✔
97	if tx['status']['confirmed']:	2✔
98	status = 'confirmed'	2✔
99	fee = None if 'fee' not in tx else tx['fee']	2✔
100	witness_type = 'legacy'	2✔
101	if tx['size'] * 4 > tx['weight']:	2✔
102	witness_type = 'segwit'	2✔
103
104	t = Transaction(locktime=tx['locktime'], version=tx['version'], network=self.network,	2✔
105	fee=fee, size=tx['size'], txid=tx['txid'],
106	date=None if 'block_time' not in tx['status'] else
107	datetime.fromtimestamp(tx['status']['block_time'], timezone.utc),
108	confirmations=confirmations, block_height=block_height, status=status,
109	coinbase=tx['vin'][0]['is_coinbase'], witness_type=witness_type)
110	index_n = 0	2✔
111	for ti in tx['vin']:	2✔
112	if tx['vin'][0]['is_coinbase']:	2✔
113	t.add_input(prev_txid=ti['txid'], output_n=ti['vout'], index_n=index_n, witness_type=witness_type,	2✔
114	unlocking_script=ti['scriptsig'], value=0, sequence=ti['sequence'], strict=self.strict)
115	else:
116	witnesses = []	2✔
117	if 'witness' in ti:	2✔
118	witnesses = [bytes.fromhex(w) for w in ti['witness']]	2✔
119	t.add_input(prev_txid=ti['txid'], output_n=ti['vout'], index_n=index_n,	2✔
120	unlocking_script=ti['scriptsig'], value=ti['prevout']['value'],
121	address='' if 'scriptpubkey_address' not in ti['prevout']
122	else ti['prevout']['scriptpubkey_address'], sequence=ti['sequence'],
123	locking_script=ti['prevout']['scriptpubkey'], witnesses=witnesses, strict=self.strict)
124	index_n += 1	2✔
125	index_n = 0	2✔
126	if len(tx['vout']) > 101:	2✔
127	# Every output needs an extra query, stop execution if there are too many transaction outputs
128	return False	1✔
129	for to in tx['vout']:	2✔
130	address = ''	2✔
131	if 'scriptpubkey_address' in to:	2✔
132	address = to['scriptpubkey_address']	2✔
133	spent = self.isspent(t.txid, index_n)	2✔
134	t.add_output(value=to['value'], address=address, lock_script=to['scriptpubkey'],	2✔
135	output_n=index_n, spent=spent, strict=self.strict)
136	index_n += 1	2✔
137	if 'segwit' in [i.witness_type for i in t.inputs] or 'p2sh-segwit' in [i.witness_type for i in t.inputs]:	2✔
138	t.witness_type = 'segwit'	2✔
139	t.update_totals()	2✔
140	t.size = tx['size']	2✔
141	return t	2✔
142
143	def gettransaction(self, txid):	2✔
144	tx = self.compose_request('tx', txid)	2✔
145	return self._parse_transaction(tx)	2✔
146
147	def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
148	self.latest_block = self.blockcount() if not self.latest_block else self.latest_block	2✔
149	prtxs = []	2✔
150	before_txid = ''	2✔
151	while True:	1✔
152	parameter = 'txs'	2✔
153	if before_txid:	2✔
154	parameter = 'txs/chain/%s' % before_txid	×
155	res = self.compose_request('address', address, parameter)	2✔
156	prtxs += res	2✔
157	if len(res) == 25:	2✔
158	before_txid = res[-1:][0]['txid']	×
159	else:
160	break	1✔
161	if len(prtxs) > limit:	×
162	break	×
163	txs = []	2✔
164	if len(set([x['status'].get('block_height', '-1') for x in prtxs])) > 1:	2✔
165	prtxs.sort(key=lambda x: x['status'].get('block_height', '-1'))	2✔
166	else:
167	prtxs = prtxs[::-1]	2✔
168	for tx in prtxs:	2✔
169	t = self._parse_transaction(tx)	2✔
170	if t:	2✔
171	txs.append(t)	2✔
172	if t.txid == after_txid:	2✔
173	txs = []	2✔
174	if len(txs) > limit:	2✔
175	break	2✔
176	return txs[:limit]	2✔
177
178	def getrawtransaction(self, txid):	2✔
179	return self.compose_request('tx', txid, 'hex')	×
180
181	def sendrawtransaction(self, rawtx):	2✔
182	res = self.compose_request('tx', post_data=rawtx, method='post')	2✔
183	return {	×
184	'txid': res,
185	'response_dict': res
186	}
187
188	def estimatefee(self, blocks):	2✔
189	est = self.compose_request('fee-estimates')	1✔
190	closest = (sorted([int(i) - blocks for i in est.keys() if int(i) - blocks >= 0]))	1✔
191	# FIXME: temporary fix for too low testnet tx fees:
192	if self.network.name == 'testnet':	1✔
193	return 2000	×
194	if closest:	1✔
195	return round(est[str(closest[0] + blocks)] * 1000)	1✔
196	else:
197	return round(est[str(sorted([int(i) for i in est.keys()])[-1:][0])] * 1000)	×
198
199	def blockcount(self):	2✔
200	return self.compose_request('blocks', 'tip', 'height')	2✔
201
202	def mempool(self, txid):	2✔
203	if txid:	2✔
204	t = self.gettransaction(txid)	2✔
205	if t and not t.confirmations:	2✔
206	return [t.txid]	×
207	else:
208	return []	2✔
209	else:
210	return self.compose_request('mempool', 'txids')	1✔
211
212	def getblock(self, blockid, parse_transactions, page, limit):	2✔
213	if isinstance(blockid, int):	1✔
214	blockid = self.compose_request('block-height', str(blockid))	×
215	if (page == 1 and limit == 10) or limit > 25:	1✔
216	limit = 25	×
217	# elif page > 1:
218	# if limit % 25 != 0:
219	# return False
220	bd = self.compose_request('block', blockid)	1✔
221	btxs = self.compose_request('block', blockid, 'txs', str((page-1)*limit))	1✔
222	if parse_transactions:	1✔
223	txs = []	1✔
224	self.latest_block = self.blockcount() if not self.latest_block else self.latest_block	1✔
225	for tx in btxs[:limit]:	1✔
226	# try:
227	txs.append(self._parse_transaction(tx))	1✔
228	# except Exception as e:
229	# _logger.error("Could not parse tx %s with error %s" % (tx['txid'], e))
230	else:
231	txs = [tx['txid'] for tx in btxs]	×
232
233	block = {	1✔
234	'bits': bd['bits'],
235	'depth': None,
236	'block_hash': bd['id'],
237	'height': bd['height'],
238	'merkle_root': bd['merkle_root'],
239	'nonce': bd['nonce'],
240	'prev_block': bd['previousblockhash'],
241	'time': bd['timestamp'],
242	'tx_count': bd['tx_count'],
243	'txs': txs,
244	'version': bd['version'],
245	'page': page,
246	'pages': None if not limit else int(bd['tx_count'] // limit) + (bd['tx_count'] % limit > 0),
247	'limit': limit
248	}
249	return block	1✔
250
251	def getrawblock(self, blockid):	2✔
252	if isinstance(blockid, int):	1✔
253	blockid = self.compose_request('block-height', str(blockid))	1✔
254	rawblock = self.compose_request('block', blockid, 'raw')	1✔
255	hexrawblock = rawblock.hex()	1✔
256	return hexrawblock	1✔
257
258	def isspent(self, txid, output_n):	2✔
259	res = self.compose_request('tx', txid, 'outspend', str(output_n))	2✔
260	return 1 if res['spent'] else 0	2✔
261
262	# 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