Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
t.block_hash = tx['blockhash']
t.fee = tx['fee']
t.rawtx = 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:
raise ClientError("Address missing in input. Provider might not support segwit transactions")
if len(t.inputs) != len(input_values):
i.value = None
continue
value = [x[1] for x in input_values if x[0] == i.address]
if len(value) != 1:
_logger.warning("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
url += url_vars
_logger.debug("Url get request %s" % url)
self.resp = requests.get(url, timeout=self.timeout, verify=secure)
elif method == 'post':
_logger.debug("Url post request %s" % url)
self.resp = requests.post(url, json=dict(variables), timeout=self.timeout, verify=secure)
resp_text = self.resp.text
if len(resp_text) > 1000:
resp_text = self.resp.text[:970] + '... truncated, length %d' % len(resp_text)
_logger.debug("Response [%d] %s" % (self.resp.status_code, resp_text))
if self.resp.status_code == 429:
raise ClientError("Maximum number of requests reached for %s with url %s, response [%d] %s" %
(self.provider, url, self.resp.status_code, resp_text))
elif not(self.resp.status_code == 200 or self.resp.status_code == 201):
raise ClientError("Error connecting to %s on url %s, response [%d] %s" %
(self.provider, url, self.resp.status_code, resp_text))
try:
return json.loads(self.resp.text)
except ValueError or json.decoder.JSONDecodeError:
return self.resp.text
def gettransactions(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
page = 0
txs = []
while True:
variables = {'mode': 'verbose', 'limit': 50, 'page': page, 'order': '1'}
try:
res = self.compose_request('address', 'transactions', address, variables)
except ClientError:
if "address not found" in self.resp.text:
return []
for tx in res['data']['list']:
txs.append(self._parse_transaction(tx))
if tx['txId'] == after_txid:
txs = []
page += 1
if page > res['data']['pages']:
break
return txs[:max_txs]
def getrawtransaction(self, tx_id):
tx = self.compose_request('rawtx', tx_id)
t = Transaction.import_raw(tx['rawtx'], 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['rawtx']
def getutxos(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
txs = []
res = self.proxy.getaddressinfo(address)
if not (res['ismine'] or res['iswatchonly']):
raise ClientError("Address %s not found in litecoind wallet, use 'importaddress' to add address to "
"wallet." % address)
for t in self.proxy.listunspent(0, 99999999, address):
txs.append({
'address': t['address'],
'tx_hash': t['txid'],
'confirmations': t['confirmations'],
'output_n': t['vout'],
'input_n': -1,
'block_height': None,
'fee': None,
'size': 0,
'value': int(t['amount'] * self.units),
'script': t['scriptPubKey'],
'date': None,
})
def gettransactions(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
txs = []
address = self._address_convert(address)
if address.witness_type != 'legacy':
raise ClientError("Provider does not support segwit addresses")
res = self.compose_request('addrs', address.address, variables={'unspentOnly': 0, 'limit': 2000})
if not isinstance(res, list):
res = [res]
for a in res:
if 'txrefs' not in a:
continue
txids = []
for t in a['txrefs'][::-1]:
if t['tx_hash'] not in txids:
txids.append(t['tx_hash'])
if t['tx_hash'] == after_txid:
txids = []
if len(txids) > 500:
_logger.warning("BlockCypher: Large number of transactions for address %s, "
"Transaction list may be incomplete" % address.address_orig)
for txid in txids[:max_txs]:
blockhash = tx['blockhash'] if 'blockhash' in tx else ''
t = Transaction(locktime=tx['locktime'], version=tx['version'], network=self.network,
fee=fees, size=tx['size'], hash=tx['txid'],
date=blocktime, confirmations=tx['confirmations'],
block_height=tx['blockheight'], block_hash=blockhash, status=status,
input_total=int(round(float(value_in) * self.units, 0)), coinbase=isCoinbase,
output_total=int(round(float(tx['valueOut']) * self.units, 0)))
for ti in tx['vin']:
# sequence = struct.pack('
def compose_request(self, func=None, path_type='api', variables=None, method='get'):
# API path: http://chainz.cryptoid.info/ltc/api.dws
# Explorer path for raw tx: https://chainz.cryptoid.info/explorer/tx.raw.dws
if variables is None:
variables = {}
if path_type == 'api':
url_path = '%s/api.dws' % self.provider_coin_id
variables.update({'q': func})
else:
url_path = 'explorer/tx.raw.dws'
variables.update({'coin': self.provider_coin_id})
if not self.api_key:
raise ClientError("Request a CryptoID API key before using this provider")
variables.update({'key': self.api_key})
return self.request(url_path, variables, method)
t.status = 'confirmed'
t.date = datetime.strptime(tx['confirmed'][:19], "%Y-%m-%dT%H:%M:%S")
else:
t.status = 'unconfirmed'
t.confirmations = tx['confirmations']
t.block_height = tx['block_height']
t.block_hash = tx.get('block_hash')
t.fee = tx['fees']
t.rawtx = tx['hex']
t.size = int(len(tx['hex']) / 2)
t.network = self.network
t.input_total = 0
if t.coinbase:
t.input_total = t.output_total
if len(t.inputs) != len(tx['inputs']):
raise ClientError("Invalid number of inputs provided. Raw tx: %d, blockcypher: %d" %
(len(t.inputs), len(tx['inputs'])))
for n, i in enumerate(t.inputs):
if not t.coinbase and not (tx['inputs'][n]['output_index'] == i.output_n_int and
tx['inputs'][n]['prev_hash'] == to_hexstring(i.prev_hash)):
raise ClientError("Transaction inputs do not match raw transaction")
if 'output_value' in tx['inputs'][n]:
i.value = tx['inputs'][n]['output_value']
t.input_total += i.value
if len(t.outputs) != len(tx['outputs']):
raise ClientError("Invalid number of outputs provided. Raw tx: %d, blockcypher: %d" %
(len(t.outputs), len(tx['outputs'])))
for n, o in enumerate(t.outputs):
if 'spent_by' in tx['outputs'][n]:
o.spent = True
t.raw_hex()
return t