1
0
mirror of https://github.com/trezor/trezor-firmware.git synced 2024-11-22 07:28:10 +00:00

trezorlib+tools: remove usage of binascii

This commit is contained in:
Pavol Rusnak 2018-09-13 00:44:08 +02:00 committed by matejcik
parent 7f55847ab1
commit 4a0ca873eb
16 changed files with 85 additions and 118 deletions

View File

@ -1,6 +1,5 @@
#!/usr/bin/env python3
import binascii
import os
import sys
@ -17,7 +16,7 @@ if os.isatty(sys.stdin.fileno()):
else:
tx_hex = sys.stdin.read().strip()
tx_bin = binascii.unhexlify(tx_hex)
tx_bin = bytes.fromhex(tx_hex)
CompactUintStruct = c.Struct(

View File

@ -11,7 +11,6 @@ import os
import sys
import json
import hashlib
import binascii
from trezorlib.client import TrezorClient
from trezorlib.transport import enumerate_devices
@ -100,7 +99,7 @@ def main():
data = {'label': label,
'bip32_path': bip32_path,
'password_encrypted_hex': binascii.hexlify(passw_encrypted).decode()}
'password_encrypted_hex': passw_encrypted.hex()}
json.dump(data, open(passw_file, 'w'))
@ -110,7 +109,7 @@ def main():
sys.stderr.write('Please confirm the action on your device ...\n')
passw = client.decrypt_keyvalue(data['bip32_path'],
data['label'],
binascii.unhexlify(data['password_encrypted_hex']),
bytes.fromhex(data['password_encrypted_hex']),
False, True)
print(passw)

View File

@ -2,7 +2,6 @@
from trezorlib.debuglink import DebugLink
from trezorlib.client import TrezorClient
from trezorlib.transport import enumerate_devices
import binascii
import sys
@ -23,7 +22,7 @@ def main():
client = TrezorClient(transport)
debug = DebugLink(debug_transport)
debug.memory_write(int(sys.argv[1], 16), binascii.unhexlify(sys.argv[2]), flash=True)
debug.memory_write(int(sys.argv[1], 16), bytes.fromhex(sys.argv[2]), flash=True)
client.close()

View File

@ -1,6 +1,6 @@
#!/usr/bin/env python3
import binascii
import hashlib
import mnemonic
__doc__ = '''
@ -48,8 +48,8 @@ def generate_entropy(strength, internal_entropy, external_entropy):
def main():
print(__doc__)
comp = binascii.unhexlify(input("Please enter computer-generated entropy (in hex): ").strip())
trzr = binascii.unhexlify(input("Please enter TREZOR-generated entropy (in hex): ").strip())
comp = bytes.fromhex(input("Please enter computer-generated entropy (in hex): ").strip())
trzr = bytes.fromhex(input("Please enter TREZOR-generated entropy (in hex): ").strip())
word_count = int(input("How many words your mnemonic has? "))
strength = word_count * 32 // 3

View File

@ -1,5 +1,4 @@
#!/usr/bin/env python3
from binascii import hexlify, unhexlify
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import hmac
@ -21,15 +20,15 @@ BIP32_PATH = parse_path("10016h/0")
def getMasterKey(client):
bip32_path = BIP32_PATH
ENC_KEY = 'Activate TREZOR Password Manager?'
ENC_VALUE = unhexlify('2d650551248d792eabf628f451200d7f51cb63e46aadcbb1038aacb05e8c8aee2d650551248d792eabf628f451200d7f51cb63e46aadcbb1038aacb05e8c8aee')
key = hexlify(client.encrypt_keyvalue(
ENC_VALUE = bytes.fromhex('2d650551248d792eabf628f451200d7f51cb63e46aadcbb1038aacb05e8c8aee2d650551248d792eabf628f451200d7f51cb63e46aadcbb1038aacb05e8c8aee')
key = client.encrypt_keyvalue(
bip32_path,
ENC_KEY,
ENC_VALUE,
True,
True
))
return key
)
return key.hex()
# Deriving file name and encryption key
@ -43,7 +42,7 @@ def getFileEncKey(key):
# File level decryption and file reading
def decryptStorage(path, key):
cipherkey = unhexlify(key)
cipherkey = bytes.fromhex(key)
with open(path, 'rb') as f:
iv = f.read(12)
tag = f.read(16)
@ -63,7 +62,7 @@ def decryptStorage(path, key):
def decryptEntryValue(nonce, val):
cipherkey = unhexlify(nonce)
cipherkey = bytes.fromhex(nonce)
iv = val[:12]
tag = val[12:28]
cipher = Cipher(algorithms.AES(cipherkey), modes.GCM(iv, tag), backend=default_backend())
@ -98,14 +97,14 @@ def getDecryptedNonce(client, entry):
ENC_KEY = 'Unlock %s for user %s?' % (item, entry['username'])
ENC_VALUE = entry['nonce']
decrypted_nonce = hexlify(client.decrypt_keyvalue(
decrypted_nonce = client.decrypt_keyvalue(
BIP32_PATH,
ENC_KEY,
unhexlify(ENC_VALUE),
bytes.fromhex(ENC_VALUE),
False,
True
))
return decrypted_nonce
)
return decrypted_nonce.hex()
# Pretty print of list
@ -163,11 +162,11 @@ def main():
pwdArr = entries[entry_id]['password']['data']
pwdHex = ''.join([hex(x)[2:].zfill(2) for x in pwdArr])
print('password: ', decryptEntryValue(plain_nonce, unhexlify(pwdHex)))
print('password: ', decryptEntryValue(plain_nonce, bytes.fromhex(pwdHex)))
safeNoteArr = entries[entry_id]['safe_note']['data']
safeNoteHex = ''.join([hex(x)[2:].zfill(2) for x in safeNoteArr])
print('safe_note:', decryptEntryValue(plain_nonce, unhexlify(safeNoteHex)))
print('safe_note:', decryptEntryValue(plain_nonce, bytes.fromhex(safeNoteHex)))
return

View File

@ -21,7 +21,6 @@
# along with this library. If not, see <http://www.gnu.org/licenses/>.
import base64
import binascii
import hashlib
import io
import json
@ -197,7 +196,7 @@ def clear_session(connect):
@click.argument("size", type=int)
@click.pass_obj
def get_entropy(connect, size):
return binascii.hexlify(misc.get_entropy(connect(), size))
return misc.get_entropy(connect(), size).hex()
@cli.command(help="Retrieve device features and settings.")
@ -538,7 +537,7 @@ def firmware_update(connect, filename, url, version, skip_check, fingerprint):
if not skip_check:
if fp[:8] == b"54525a52" or fp[:8] == b"54525a56":
fp = binascii.unhexlify(fp)
fp = bytes.fromhex(fp.decode())
if fp[:4] != b"TRZR" and fp[:4] != b"TRZV":
click.echo("Trezor firmware header expected.")
sys.exit(2)
@ -618,8 +617,8 @@ def get_public_node(connect, coin, address, curve, script_type, show_display):
"depth": result.node.depth,
"fingerprint": "%08x" % result.node.fingerprint,
"child_num": result.node.child_num,
"chain_code": binascii.hexlify(result.node.chain_code),
"public_key": binascii.hexlify(result.node.public_key),
"chain_code": result.node.chain_code.hex(),
"public_key": result.node.public_key.hex(),
},
"xpub": result.xpub,
}
@ -662,7 +661,7 @@ def sign_tx(connect, coin):
def outpoint(s):
txid, vout = s.split(":")
return binascii.unhexlify(txid), int(vout)
return bytes.fromhex(txid), int(vout)
inputs = []
while True:
@ -691,9 +690,7 @@ def sign_tx(connect, coin):
else CHOICE_INPUT_SCRIPT_TYPE.typemap[script_type]
)
if txapi.bip115:
prev_output = txapi.get_tx(
binascii.hexlify(prev_hash).decode()
).bin_outputs[prev_index]
prev_output = txapi.get_tx(prev_hash.hex()).bin_outputs[prev_index]
prev_blockhash = prev_output.block_hash
prev_blockheight = prev_output.block_height
inputs.append(
@ -762,7 +759,7 @@ def sign_tx(connect, coin):
click.echo()
click.echo("Signed Transaction:")
click.echo(binascii.hexlify(serialized_tx))
click.echo(serialized_tx.hex())
click.echo()
click.echo("Use the following form to broadcast it to the network:")
click.echo(txapi.pushtx_url)
@ -826,17 +823,17 @@ def ethereum_sign_message(connect, address, message):
ret = ethereum.sign_message(client, address_n, message)
output = {
"message": message,
"address": "0x%s" % binascii.hexlify(ret.address).decode(),
"signature": "0x%s" % binascii.hexlify(ret.signature).decode(),
"address": "0x%s" % ret.address.hex(),
"signature": "0x%s" % ret.signature.hex(),
}
return output
def ethereum_decode_hex(value):
if value.startswith("0x") or value.startswith("0X"):
return binascii.unhexlify(value[2:])
return bytes.fromhex(value[2:])
else:
return binascii.unhexlify(value)
return bytes.fromhex(value)
@cli.command(help="Verify message signed with Ethereum address.")
@ -859,7 +856,7 @@ def encrypt_keyvalue(connect, address, key, value):
client = connect()
address_n = tools.parse_path(address)
res = misc.encrypt_keyvalue(client, address_n, key, value.encode())
return binascii.hexlify(res)
return res.hex()
@cli.command(help="Decrypt value by given key and path.")
@ -870,7 +867,7 @@ def encrypt_keyvalue(connect, address, key, value):
def decrypt_keyvalue(connect, address, key, value):
client = connect()
address_n = tools.parse_path(address)
return misc.decrypt_keyvalue(client, address_n, key, binascii.unhexlify(value))
return misc.decrypt_keyvalue(client, address_n, key, bytes.fromhex(value))
# @cli.command(help='Encrypt message.')
@ -882,13 +879,13 @@ def decrypt_keyvalue(connect, address, key, value):
# @click.pass_obj
# def encrypt_message(connect, coin, display_only, address, pubkey, message):
# client = connect()
# pubkey = binascii.unhexlify(pubkey)
# pubkey = bytes.fromhex(pubkey)
# address_n = tools.parse_path(address)
# res = client.encrypt_message(pubkey, message, display_only, coin, address_n)
# return {
# 'nonce': binascii.hexlify(res.nonce),
# 'message': binascii.hexlify(res.message),
# 'hmac': binascii.hexlify(res.hmac),
# 'nonce': res.nonce.hex(),
# 'message': res.message.hex(),
# 'hmac': res.hmac.hex(),
# 'payload': base64.b64encode(res.nonce + res.message + res.hmac),
# }
@ -920,7 +917,7 @@ def ethereum_get_address(connect, address, show_display):
client = connect()
address_n = tools.parse_path(address)
address = ethereum.get_address(client, address_n, show_display)
return "0x%s" % binascii.hexlify(address).decode()
return "0x%s" % address.hex()
@cli.command(
@ -1027,9 +1024,7 @@ def ethereum_sign_tx(
client = connect()
address_n = tools.parse_path(address)
address = "0x%s" % (
binascii.hexlify(ethereum.get_address(client, address_n)).decode()
)
address = "0x%s" % ethereum.get_address(client, address_n).hex()
if gas_price is None or gas_limit is None or nonce is None or publish:
host, port = host.split(":")
@ -1047,7 +1042,7 @@ def ethereum_sign_tx(
to_address=to,
from_address=address,
value=("0x%x" % value),
data="0x%s" % (binascii.hexlify(data).decode()),
data="0x%s" % data.hex(),
)
if nonce is None:
@ -1074,7 +1069,7 @@ def ethereum_sign_tx(
transaction = rlp.encode(
(tx_type, nonce, gas_price, gas_limit, to_address, value, data) + sig
)
tx_hex = "0x%s" % binascii.hexlify(transaction).decode()
tx_hex = "0x%s" % transaction.hex()
if publish:
tx_hash = eth.eth_sendRawTransaction(tx_hex)
@ -1110,8 +1105,8 @@ def cardano_sign_tx(connect, file, network):
signed_transaction = cardano.sign_tx(client, inputs, outputs, transactions, network)
return {
"tx_hash": binascii.hexlify(signed_transaction.tx_hash).decode(),
"tx_body": binascii.hexlify(signed_transaction.tx_body).decode(),
"tx_hash": signed_transaction.tx_hash.hex(),
"tx_body": signed_transaction.tx_body.hex(),
}
@ -1174,10 +1169,7 @@ def nem_sign_tx(connect, address, file, broadcast):
address_n = tools.parse_path(address)
transaction = nem.sign_tx(client, address_n, json.load(file))
payload = {
"data": binascii.hexlify(transaction.data).decode(),
"signature": binascii.hexlify(transaction.signature).decode(),
}
payload = {"data": transaction.data.hex(), "signature": transaction.signature.hex()}
if broadcast:
import requests
@ -1216,7 +1208,7 @@ def lisk_get_public_key(connect, address, show_display):
client = connect()
address_n = tools.parse_path(address)
res = lisk.get_public_key(client, address_n, show_display)
output = {"public_key": binascii.hexlify(res.public_key).decode()}
output = {"public_key": res.public_key.hex()}
return output
@ -1237,7 +1229,7 @@ def lisk_sign_tx(connect, address, file):
address_n = tools.parse_path(address)
transaction = lisk.sign_tx(client, address_n, json.load(file))
payload = {"signature": binascii.hexlify(transaction.signature).decode()}
payload = {"signature": transaction.signature.hex()}
return payload
@ -1254,8 +1246,8 @@ def lisk_sign_message(connect, address, message):
res = lisk.sign_message(client, address_n, message)
output = {
"message": message,
"public_key": binascii.hexlify(res.public_key).decode(),
"signature": binascii.hexlify(res.signature).decode(),
"public_key": res.public_key.hex(),
"signature": res.signature.hex(),
}
return output
@ -1285,7 +1277,7 @@ def lisk_verify_message(connect, pubkey, signature, message):
def cosi_commit(connect, address, data):
client = connect()
address_n = tools.parse_path(address)
return cosi.commit(client, address_n, binascii.unhexlify(data))
return cosi.commit(client, address_n, bytes.fromhex(data))
@cli.command(help="Ask device to sign using CoSi.")
@ -1302,9 +1294,9 @@ def cosi_sign(connect, address, data, global_commitment, global_pubkey):
return cosi.sign(
client,
address_n,
binascii.unhexlify(data),
binascii.unhexlify(global_commitment),
binascii.unhexlify(global_pubkey),
bytes.fromhex(data),
bytes.fromhex(global_commitment),
bytes.fromhex(global_pubkey),
)
@ -1383,10 +1375,10 @@ def ripple_sign_tx(connect, address, file):
result = ripple.sign_tx(client, address_n, msg)
click.echo("Signature:")
click.echo(binascii.hexlify(result.signature))
click.echo(result.signature.hex())
click.echo()
click.echo("Serialized tx including the signature:")
click.echo(binascii.hexlify(result.serialized_tx))
click.echo(result.serialized_tx.hex())
#
@ -1457,7 +1449,7 @@ def ontology_get_public_key(connect, address, show_display):
address_n = tools.parse_path(address)
result = ontology.get_public_key(client, address_n, show_display)
return binascii.hexlify(result.public_key).decode()
return result.public_key.hex()
@cli.command(help="Sign Ontology transfer.")
@ -1492,10 +1484,7 @@ def ontology_sign_transfer(connect, address, transaction_f, transfer_f):
result = ontology.sign_transfer(client, address_n, transaction, transfer)
output = {
"payload": binascii.hexlify(result.payload).decode(),
"signature": binascii.hexlify(result.signature).decode(),
}
output = {"payload": result.payload.hex(), "signature": result.signature.hex()}
return output
@ -1534,10 +1523,7 @@ def ontology_sign_withdraw_ong(connect, address, transaction_f, withdraw_ong_f):
result = ontology.sign_withdrawal(client, address_n, transaction, withdraw_ong)
output = {
"payload": binascii.hexlify(result.payload).decode(),
"signature": binascii.hexlify(result.signature).decode(),
}
output = {"payload": result.payload.hex(), "signature": result.signature.hex()}
return output
@ -1578,10 +1564,7 @@ def ontology_sign_ont_id_register(connect, address, transaction_f, ont_id_regist
result = ontology.sign_register(client, address_n, transaction, ont_id_register)
output = {
"payload": binascii.hexlify(result.payload).decode(),
"signature": binascii.hexlify(result.signature).decode(),
}
output = {"payload": result.payload.hex(), "signature": result.signature.hex()}
return output
@ -1624,10 +1607,7 @@ def ontology_sign_ont_id_add_attributes(
client, address_n, transaction, ont_id_add_attributes
)
output = {
"payload": binascii.hexlify(result.payload).decode(),
"signature": binascii.hexlify(result.signature).decode(),
}
output = {"payload": result.payload.hex(), "signature": result.signature.hex()}
return output

View File

@ -14,7 +14,6 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import binascii
from typing import List
from . import messages, tools
@ -56,7 +55,7 @@ def sign_tx(
while isinstance(response, messages.CardanoTxRequest):
tx_index = response.tx_index
transaction_data = binascii.unhexlify(transactions[tx_index])
transaction_data = bytes.fromhex(transactions[tx_index])
ack_message = messages.CardanoTxAck(transaction=transaction_data)
response = client.call(ack_message)
@ -71,7 +70,7 @@ def create_input(input) -> messages.CardanoTxInputType:
return messages.CardanoTxInputType(
address_n=tools.parse_path(path),
prev_hash=binascii.unhexlify(input["prev_hash"]),
prev_hash=bytes.fromhex(input["prev_hash"]),
prev_index=input["prev_index"],
type=input["type"],
)

View File

@ -14,7 +14,6 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import binascii
import functools
import getpass
import logging
@ -498,7 +497,7 @@ class ProtocolMixin(object):
if not self.tx_api:
raise RuntimeError("TX_API not defined")
prev_tx = self.tx_api.get_tx(binascii.hexlify(inp.prev_hash).decode())
prev_tx = self.tx_api.get_tx(inp.prev_hash.hex())
txes[inp.prev_hash] = prev_tx
return txes

View File

@ -14,7 +14,6 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import binascii
from functools import reduce
from typing import Iterable, Tuple
@ -63,7 +62,7 @@ def get_nonce(
r = _ed25519.Hint(
bytes([h[i] for i in range(b >> 3, b >> 2)])
+ data
+ binascii.unhexlify("%08x" % ctr)
+ bytes.fromhex("%08x" % ctr)
)
R = _ed25519.scalarmult(_ed25519.B, r)
return r, Ed25519PublicPoint(_ed25519.encodepoint(R))

View File

@ -14,7 +14,6 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import binascii
from mnemonic import Mnemonic
@ -204,12 +203,12 @@ def load_device_by_xprv(client, xprv, pin, passphrase_protection, label, languag
raise ValueError("Invalid length of xprv")
node = proto.HDNodeType()
data = binascii.hexlify(tools.b58decode(xprv, None))
data = tools.b58decode(xprv, None).hex()
if data[90:92] != b"00":
if data[90:92] != "00":
raise ValueError("Contain invalid private key")
checksum = binascii.hexlify(tools.btc_hash(binascii.unhexlify(data[:156]))[:4])
checksum = (tools.btc_hash(bytes.fromhex(data[:156]))[:4]).hex()
if checksum != data[156:]:
raise ValueError("Checksum doesn't match")
@ -224,8 +223,8 @@ def load_device_by_xprv(client, xprv, pin, passphrase_protection, label, languag
node.depth = int(data[8:10], 16)
node.fingerprint = int(data[10:18], 16)
node.child_num = int(data[18:26], 16)
node.chain_code = binascii.unhexlify(data[26:90])
node.private_key = binascii.unhexlify(data[92:156]) # skip 0x00 indicating privkey
node.chain_code = bytes.fromhex(data[26:90])
node.private_key = bytes.fromhex(data[92:156]) # skip 0x00 indicating privkey
resp = client.call(
proto.LoadDevice(

View File

@ -177,7 +177,7 @@ def reset(
raise RuntimeError("Invalid response, expected EntropyRequest")
external_entropy = os.urandom(32)
# LOG.debug("Computer generated entropy: " + binascii.hexlify(external_entropy).decode())
# LOG.debug("Computer generated entropy: " + external_entropy.hex())
ret = client.call(proto.EntropyAck(entropy=external_entropy))
client.init_device()
return ret

View File

@ -1,4 +1,3 @@
import binascii
import construct as c
import pyblake2
@ -107,7 +106,7 @@ def validate_firmware(filename):
with open(filename, "rb") as f:
data = f.read()
if data[:6] == b"54525a":
data = binascii.unhexlify(data)
data = bytes.fromhex(data.decode())
try:
fw = Firmware.parse(data)
@ -135,7 +134,7 @@ def validate_firmware(filename):
header_bytes = FirmwareHeader.build(stripped_header)
digest = pyblake2.blake2s(header_bytes).digest()
print("Fingerprint: {}".format(binascii.hexlify(digest).decode()))
print("Fingerprint: {}".format(digest.hex()))
global_pk = cosi.combine_keys(
vendor.pubkeys[i] for i in range(8) if header.sigmask & (1 << i)

View File

@ -14,7 +14,6 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import binascii
import json
from . import messages as proto
@ -38,7 +37,7 @@ def create_transaction_common(transaction):
msg.deadline = transaction["deadline"]
if "signer" in transaction:
msg.signer = binascii.unhexlify(transaction["signer"])
msg.signer = bytes.fromhex(transaction["signer"])
return msg
@ -49,10 +48,10 @@ def create_transfer(transaction):
msg.amount = transaction["amount"]
if "payload" in transaction["message"]:
msg.payload = binascii.unhexlify(transaction["message"]["payload"])
msg.payload = bytes.fromhex(transaction["message"]["payload"])
if transaction["message"]["type"] == 0x02:
msg.public_key = binascii.unhexlify(transaction["message"]["publicKey"])
msg.public_key = bytes.fromhex(transaction["message"]["publicKey"])
if "mosaics" in transaction:
msg.mosaics = [
@ -72,7 +71,7 @@ def create_aggregate_modification(transactions):
msg.modifications = [
proto.NEMCosignatoryModification(
type=modification["modificationType"],
public_key=binascii.unhexlify(modification["cosignatoryAccount"]),
public_key=bytes.fromhex(modification["cosignatoryAccount"]),
)
for modification in transactions["modifications"]
]
@ -141,7 +140,7 @@ def create_supply_change(transaction):
def create_importance_transfer(transaction):
msg = proto.NEMImportanceTransfer()
msg.mode = transaction["importanceTransfer"]["mode"]
msg.public_key = binascii.unhexlify(transaction["importanceTransfer"]["publicKey"])
msg.public_key = bytes.fromhex(transaction["importanceTransfer"]["publicKey"])
return msg

View File

@ -37,7 +37,6 @@ required:
>>> """
'''
import binascii
from io import BytesIO
from typing import Any, Optional
@ -391,7 +390,7 @@ def format_message(
if mostly_printable(value):
output = repr(value)
else:
output = "0x" + binascii.hexlify(value).decode()
output = "0x" + value.hex()
return "{} bytes {}{}".format(length, output, suffix)
return repr(value)
@ -418,7 +417,7 @@ def value_to_proto(ftype, value):
if ftype is BytesType:
if isinstance(value, str):
return binascii.unhexlify(value)
return bytes.fromhex(value)
elif isinstance(value, bytes):
return value
else:

View File

@ -14,7 +14,6 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import binascii
import logging
import struct
from io import BytesIO
@ -96,7 +95,7 @@ class BridgeTransport(Transport):
protobuf.dump_message(data, msg)
ser = data.getvalue()
header = struct.pack(">HL", mapping.get_type(msg), len(ser))
data = binascii.hexlify(header + ser).decode()
data = (header + ser).hex()
r = self.conn.post(
TREZORD_HOST + "/call/%s" % self.session, data=data, headers=self.HEADERS
)
@ -107,7 +106,7 @@ class BridgeTransport(Transport):
def read(self):
if self.response is None:
raise TransportException("No response stored")
data = binascii.unhexlify(self.response)
data = bytes.fromhex(self.response)
headerlen = struct.calcsize(">HL")
(msg_type, datalen) = struct.unpack(">HL", data[:headerlen])
data = BytesIO(data[headerlen : headerlen + datalen])

View File

@ -14,7 +14,6 @@
# You should have received a copy of the License along with this library.
# If not, see <https://www.gnu.org/licenses/lgpl-3.0.html>.
import binascii
import json
from decimal import Decimal
@ -78,7 +77,7 @@ class TxApiInsight(TxApi):
def get_block_hash(self, block_number):
j = self.fetch_json("block-index", block_number)
return binascii.unhexlify(j["blockHash"])
return bytes.fromhex(j["blockHash"])
def current_height(self):
r = requests.get(self.get_url("status?q=getBlockCount"))
@ -99,19 +98,19 @@ class TxApiInsight(TxApi):
if "coinbase" in vin.keys():
i.prev_hash = b"\0" * 32
i.prev_index = 0xffffffff # signed int -1
i.script_sig = binascii.unhexlify(vin["coinbase"])
i.script_sig = bytes.fromhex(vin["coinbase"])
i.sequence = vin["sequence"]
else:
i.prev_hash = binascii.unhexlify(vin["txid"])
i.prev_hash = bytes.fromhex(vin["txid"])
i.prev_index = vin["vout"]
i.script_sig = binascii.unhexlify(vin["scriptSig"]["hex"])
i.script_sig = bytes.fromhex(vin["scriptSig"]["hex"])
i.sequence = vin["sequence"]
for vout in data["vout"]:
o = t._add_bin_outputs()
o.amount = int(Decimal(vout["value"]) * 100000000)
o.script_pubkey = binascii.unhexlify(vout["scriptPubKey"]["hex"])
o.script_pubkey = bytes.fromhex(vout["scriptPubKey"]["hex"])
if self.bip115 and o.script_pubkey[-1] == 0xb4:
# Verify if coin implements replay protection bip115 and script includes checkblockatheight opcode. 0xb4 - is op_code (OP_CHECKBLOCKATHEIGHT)
# <OP_32> <32-byte block hash> <OP_3> <3-byte block height> <OP_CHECKBLOCKATHEIGHT>
@ -134,7 +133,7 @@ class TxApiInsight(TxApi):
raise ValueError("Too many joinsplits")
extra_data_len = 1 + joinsplit_cnt * 1802 + 32 + 64
raw = self.fetch_json("rawtx", txhash)
raw = binascii.unhexlify(raw["rawtx"])
raw = bytes.fromhex(raw["rawtx"])
t.extra_data = raw[-extra_data_len:]
return t