import json from struct import unpack, pack from math import ceil from io import BytesIO from pybtc.constants import * from pybtc.opcodes import * from pybtc.functions.tools import (int_to_var_int, read_var_int, var_int_to_int, rh2s, s2rh, bytes_from_hex, get_stream) from pybtc.functions.script import op_push_data, decode_script, parse_script, sign_message from pybtc.functions.script import get_multisig_public_keys, read_opcode, is_valid_signature_encoding from pybtc.functions.script import public_key_recovery, delete_from_script from pybtc.functions.hash import hash160, sha256, double_sha256 from pybtc.functions.address import hash_to_address, address_net_type, address_to_script from pybtc.address import PrivateKey, Address, ScriptAddress, PublicKey class Transaction(dict): """ The class for Transaction object :param raw_tx: (optional) raw transaction in bytes or HEX encoded string, if no raw transaction provided well be created new empty transaction template. :param tx_format: "raw" or "decoded" format. Raw format is mean that all transaction represented in bytes for best performance. Decoded transaction is represented in human readable format using base68, hex, bech32, asm and opcodes. By default "decoded" format using. :param int version: transaction version for new template, by default 1. :param int lock_time: transaction lock time for new template, by default 0. :param boolean testnet: address type for "decoded" transaction representation. """ def __init__(self, raw_tx=None, format="decoded", version=1, lock_time=0, testnet=False, auto_commit=True): if format not in ("decoded", "raw"): raise ValueError("format error, raw or decoded allowed") self.auto_commit = auto_commit self["format"] = format self["testnet"] = testnet self["segwit"] = False self["txId"] = None self["hash"] = None self["version"] = version self["size"] = 0 self["vSize"] = 0 self["bSize"] = 0 self["lockTime"] = lock_time self["vIn"] = dict() self["vOut"] = dict() self["rawTx"] = None self["blockHash"] = None self["confirmations"] = None self["time"] = None self["blockTime"] = None self["blockIndex"] = None self["coinbase"] = False self["fee"] = None self["data"] = None self["amount"] = None if raw_tx is None: return self["amount"] = 0 sw = sw_len = 0 stream = self.get_stream(raw_tx) start = stream.tell() read = stream.read tell = stream.tell seek = stream.seek # start deserialization self["version"] = unpack(' 520 and input_verify): raise TypeError("script_sig invalid") if not isinstance(v_out, int) or not (v_out <= 0xffffffff and v_out >= 0): raise TypeError("v_out invalid") if not isinstance(sequence, int) or not (sequence <= 0xffffffff and sequence >= 0): raise TypeError("sequence invalid") if private_key: if not isinstance(private_key, PrivateKey): private_key = PrivateKey(private_key) if amount: if not isinstance(amount, int) or not amount >= 0 and amount <= MAX_AMOUNT: raise TypeError("amount invalid") if tx_in_witness: if not isinstance(tx_in_witness, list): raise TypeError("tx_in_witness invalid") l = 0 witness = [] for w in tx_in_witness: if isinstance(w, str): witness.append(bytes_from_hex(w) if self["format"] == "raw" else w) else: witness.append(w if self["format"] == "raw" else bytes_from_hex(w)) l += 1 + len(w) if len(w) >= 0x4c: l += 1 if len(w) > 0xff: l += 1 # witness script limit if not l <= 10000: raise TypeError("tx_in_witness invalid") if tx_id == b"\x00" * 32: if not (v_out == 0xffffffff and sequence == 0xffffffff and len(script_sig) <= 100): if input_verify: raise TypeError("coinbase tx invalid") self["coinbase"] = True # script_pub_key if script_pub_key: if isinstance(script_pub_key, str): script_pub_key = bytes_from_hex(script_pub_key) if not isinstance(script_pub_key, bytes): raise TypeError("script_pub_key tx invalid") if redeem_script: if isinstance(redeem_script, str): redeem_script = bytes_from_hex(redeem_script) if not isinstance(redeem_script, bytes): raise TypeError("redeem_script tx invalid") if address is not None: if isinstance(address, str): net = True if address_net_type(address) == 'mainnet' else False if not net != self["testnet"]: raise TypeError("address invalid") script = address_to_script(address) elif type(address) in (Address, ScriptAddress): script = address_to_script(address.address) else: raise TypeError("address invalid") if script_pub_key: if script_pub_key != script: raise Exception("address not match script") else: script_pub_key = script k = len(self["vIn"]) self["vIn"][k] = dict() self["vIn"][k]["vOut"] = v_out self["vIn"][k]["sequence"] = sequence if self["format"] == "raw": self["vIn"][k]["txId"] = tx_id self["vIn"][k]["scriptSig"] = script_sig if script_pub_key: self["vIn"][k]["scriptPubKey"] = script_pub_key if redeem_script: self["vIn"][k]["redeemScript"] = redeem_script else: self["vIn"][k]["txId"] = rh2s(tx_id) self["vIn"][k]["scriptSig"] = script_sig.hex() self["vIn"][k]["scriptSigOpcodes"] = decode_script(script_sig) self["vIn"][k]["scriptSigAsm"] = decode_script(script_sig, 1) if script_pub_key: self["vIn"][k]["scriptPubKey"] = script_pub_key.hex() self["vIn"][k]["scriptPubKeyOpcodes"] = decode_script(script_pub_key) self["vIn"][k]["scriptPubKeyAsm"] = decode_script(script_pub_key, 1) if redeem_script: self["vIn"][k]["redeemScript"] = redeem_script.hex() self["vIn"][k]["redeemScriptOpcodes"] = decode_script(redeem_script) self["vIn"][k]["redeemScriptAsm"] = decode_script(script_pub_key, 1) if tx_in_witness: self["segwit"] = True self["vIn"][k]["txInWitness"] = witness if amount: self["vIn"][k]["value"] = amount if private_key: self["vIn"][k]["private_key"] = private_key if self.auto_commit: self.commit() return self def add_output(self, amount, address=None, script_pub_key=None): if address is None and script_pub_key is None: raise Exception("unable to add output, address or script required") if type(amount) != int: raise TypeError("unable to add output, amount type error") if amount < 0 or amount > MAX_AMOUNT: raise Exception("unable to add output, amount value error") if script_pub_key: if isinstance(script_pub_key, str): script_pub_key = bytes_from_hex(script_pub_key) if not isinstance(script_pub_key, bytes): raise TypeError("unable to add output, script_pub_key type error") else: if type(address) == Address: address = address.address script_pub_key = address_to_script(address) k = len(self["vOut"]) self["vOut"][k] = dict() self["vOut"][k]["value"] = amount segwit = True if "segwit" in self else False s = parse_script(script_pub_key, segwit) self["vOut"][k]["nType"] = s["nType"] self["vOut"][k]["type"] = s["type"] if self["format"] == "raw": self["vOut"][k]["scriptPubKey"] = script_pub_key if self["data"] is None: if s["nType"] == 3: self["data"] = s["data"] if s["nType"] not in (3, 4, 7, 8): self["vOut"][k]["addressHash"] = s["addressHash"] self["vOut"][k]["reqSigs"] = s["reqSigs"] else: self["vOut"][k]["scriptPubKey"] = script_pub_key.hex() if self["data"] is None: if s["nType"] == 3: self["data"] = s["data"].hex() if s["nType"] not in (3, 4, 7, 8): self["vOut"][k]["addressHash"] = s["addressHash"].hex() self["vOut"][k]["reqSigs"] = s["reqSigs"] self["vOut"][k]["scriptPubKeyOpcodes"] = decode_script(script_pub_key) self["vOut"][k]["scriptPubKeyAsm"] = decode_script(script_pub_key, 1) sh = True if self["vOut"][k]["nType"] in (1, 5) else False witness_version = None if self["vOut"][k]["nType"] < 5 else 0 if "addressHash" in self["vOut"][k]: self["vOut"][k]["address"] = hash_to_address(self["vOut"][k]["addressHash"], self["testnet"], sh, witness_version) if self.auto_commit: self.commit() return self def del_output(self, n=None): if not self["vOut"]: return self if n is None: n = len(self["vOut"]) - 1 new_out = dict() c = 0 for i in range(len(self["vOut"])): if i != n: new_out[c] = self["vOut"][i] c += 1 self["vOut"] = new_out if self.auto_commit: self.commit() return self def del_input(self, n): if not self["vIn"]: return self if n is None: n = len(self["vIn"]) - 1 new_in = dict() c = 0 for i in range(len(self["vIn"])): if i != n: new_in[c] = self["vIn"][i] c += 1 self["vIn"] = new_in if self.auto_commit: self.commit() return self def sign_input(self, n, private_key=None, script_pub_key=None, redeem_script=None, sighash_type=SIGHASH_ALL, address=None, amount=None, witness_version=0, p2sh_p2wsh=False): # private key if not private_key: try: private_key = self["vIn"][n]["private_key"].key except: raise RuntimeError("no private key") if isinstance(private_key, list): public_key = [PublicKey(p).key for p in private_key] private_key = [p.key if isinstance(p, PrivateKey) else PrivateKey(p).key for p in private_key] else: if not isinstance(private_key, PrivateKey): private_key = PrivateKey(private_key) public_key = [PublicKey(private_key).key] private_key = [private_key.key] if address is not None: if isinstance(address, str): net = True if address_net_type(address) == 'mainnet' else False if not net != self["testnet"]: raise TypeError("address invalid") script_pub_key = address_to_script(address) elif type(address) in (Address, ScriptAddress): script_pub_key = address_to_script(address.address) # script pub key if script_pub_key is None: if "scriptPubKey" in self["vIn"][n]: script_pub_key = self["vIn"][n]["scriptPubKey"] st = parse_script(script_pub_key) elif redeem_script or "redeemScript" in self["vIn"][n]: if witness_version is None or p2sh_p2wsh: st = {"type": "P2SH"} elif witness_version == 0: st = {"type": "P2WSH"} else: raise RuntimeError("not implemented") else: raise RuntimeError("no scriptPubKey key") else: st = parse_script(script_pub_key) if isinstance(script_pub_key, str): script_pub_key = bytes.fromhex(script_pub_key) # sign input if st["type"] == "PUBKEY": script_sig = self.__sign_pubkey__(n, private_key, script_pub_key, sighash_type) elif st["type"] == "P2PKH": script_sig = self.__sign_p2pkh__(n, private_key, public_key, script_pub_key, sighash_type) elif st["type"] == "P2SH": script_sig = self.__sign_p2sh(n, private_key, public_key, redeem_script, sighash_type, amount, p2sh_p2wsh) elif st["type"] == "P2WPKH": script_sig = self.__sign_p2wpkh(n, private_key, public_key, script_pub_key, sighash_type, amount) elif st["type"] == "P2WSH": script_sig = self.__sign_p2wsh(n, private_key, public_key, script_pub_key, redeem_script, sighash_type, amount) elif st["type"] == "MULTISIG": script_sig = self.__sign_bare_multisig__(n, private_key, public_key, script_pub_key, sighash_type) else: raise RuntimeError("not implemented") if self["format"] == "raw": self["vIn"][n]["scriptSig"] = script_sig else: self["vIn"][n]["scriptSig"] = script_sig.hex() self["vIn"][n]["scriptSigOpcodes"] = decode_script(script_sig) self["vIn"][n]["scriptSigAsm"] = decode_script(script_sig, 1) if self.auto_commit: self.commit() return self def __sign_bare_multisig__(self, n, private_key, public_key, script_pub_key, sighash_type): sighash = self.sig_hash(n, script_pub_key=script_pub_key, sighash_type=sighash_type) sighash = s2rh(sighash) if isinstance(sighash, str) else sighash sig = [sign_message(sighash, p, 0) + bytes([sighash_type]) for p in private_key] self["vIn"][n]['signatures'] = [s if self["format"] == "raw" else s.hex() for s in sig] return b''.join(self.__get_bare_multisig_script_sig__(self["vIn"][n]["scriptSig"], script_pub_key, public_key, sig, n)) def __sign_pubkey__(self, n, private_key, script_pub_key, sighash_type): sighash = self.sig_hash(n, script_pub_key=script_pub_key, sighash_type=sighash_type) sighash = s2rh(sighash) if isinstance(sighash, str) else sighash signature = sign_message(sighash, private_key[0], 0) + bytes([sighash_type]) self["vIn"][n]['signatures'] = [signature, ] if self["format"] == "raw" else [signature.hex(), ] return b''.join([bytes([len(signature)]), signature]) def __sign_p2pkh__(self, n, private_key, public_key, script_pub_key, sighash_type): sighash = self.sig_hash(n, script_pub_key=script_pub_key, sighash_type=sighash_type) sighash = s2rh(sighash) if isinstance(sighash, str) else sighash signature = sign_message(sighash, private_key[0], 0) + bytes([sighash_type]) self["vIn"][n]['signatures'] = [signature, ] if self["format"] == "raw" else [signature.hex(), ] script_sig = b''.join([bytes([len(signature)]), signature, bytes([len(public_key[0])]), public_key[0]]) return script_sig def __sign_p2sh(self, n, private_key, public_key, redeem_script, sighash_type, amount, p2sh_p2wsh): if not redeem_script: if "redeemScript" in self["vIn"][n]: redeem_script = self["vIn"][n]["redeemScript"] else: raise RuntimeError("no redeem script") if isinstance(redeem_script, str): redeem_script = bytes.fromhex(redeem_script) rst = parse_script(redeem_script) if p2sh_p2wsh: return self.__sign_p2sh_p2wsh(n, private_key, public_key, redeem_script, sighash_type, amount) if rst["type"] == "MULTISIG": return self.__sign_p2sh_multisig(n, private_key, public_key, redeem_script, sighash_type) elif rst["type"] == "P2WPKH": return self.__sign_p2sh_p2wpkh(n, private_key, public_key, redeem_script, sighash_type, amount) else: return self.__sign_p2sh_custom(n, private_key, public_key, redeem_script, sighash_type, amount) def __sign_p2sh_multisig(self, n, private_key, public_key, redeem_script, sighash_type): sighash = self.sig_hash(n, script_pub_key=redeem_script, sighash_type=sighash_type) sighash = s2rh(sighash) if isinstance(sighash, str) else sighash sig = [sign_message(sighash, p, 0) + bytes([sighash_type]) for p in private_key] self["vIn"][n]['signatures'] = [s if self["format"] == "raw" else s.hex() for s in sig] return b''.join(self.__get_multisig_script_sig__(self["vIn"][n]["scriptSig"], public_key, sig, redeem_script, redeem_script, n)) def __sign_p2sh_p2wpkh(self, n, private_key, public_key, redeem_script, sighash_type, amount): s = [b'\x19', OP_DUP, OP_HASH160, op_push_data(hash160(public_key[0], 0)), OP_EQUALVERIFY, OP_CHECKSIG] if amount is None: try: amount = self["vIn"][n]["value"] except: raise RuntimeError("no input amount") sighash = self.sig_hash_segwit(n, amount, script_pub_key=b"".join(s), sighash_type=sighash_type) sighash = bytes.fromhex(sighash) if isinstance(sighash, str) else sighash signature = sign_message(sighash, private_key[0], 0) + bytes([sighash_type]) self["segwit"] = True if self["format"] == "raw": self["vIn"][n]['txInWitness'] = [signature, public_key[0]] else: self["vIn"][n]['txInWitness'] = [signature.hex(), public_key[0].hex()] self["vIn"][n]['signatures'] = [signature,] if self["format"] == "raw" else [signature.hex(),] return op_push_data(redeem_script) def __sign_p2sh_p2wsh(self, n, private_key, public_key, redeem_script, sighash_type, amount): rst = parse_script(redeem_script) if rst["type"] == "MULTISIG": return self.__sign_p2sh_p2wsh_multisig(n, private_key, public_key, redeem_script, sighash_type, amount) else: raise RuntimeError("not implemented") def __sign_p2sh_custom(self, n, private_key, public_key, redeem_script, sighash_type, amount): raise RuntimeError("not implemented") return b"" def __sign_p2wpkh(self, n, private_key, public_key, script_pub_key, sighash_type, amount): s = [b'\x19', OP_DUP, OP_HASH160, script_pub_key[1:], OP_EQUALVERIFY, OP_CHECKSIG] if amount is None: try: amount = self["vIn"][n]["value"] except: raise RuntimeError("no input amount") sighash = self.sig_hash_segwit(n, amount, script_pub_key=b"".join(s), sighash_type=sighash_type) sighash = bytes.fromhex(sighash) if isinstance(sighash, str) else sighash signature = sign_message(sighash, private_key[0], 0) + bytes([sighash_type]) self["segwit"] = True if self["format"] == "raw": self["vIn"][n]['txInWitness'] = [signature, public_key[0]] else: self["vIn"][n]['txInWitness'] = [signature.hex(), public_key[0].hex()] self["vIn"][n]['signatures'] = [signature,] if self["format"] == "raw" else [signature.hex(),] return b"" def __sign_p2wsh(self, n, private_key, public_key, script_pub_key, redeem_script, sighash_type, amount): self["segwit"] = True if not redeem_script: if "redeemScript" in self["vIn"][n]: redeem_script = self["vIn"][n]["redeemScript"] else: raise RuntimeError("no redeem script") if isinstance(redeem_script, str): redeem_script = bytes.fromhex(redeem_script) rst = parse_script(redeem_script) if amount is None: try: amount = self["vIn"][n]["value"] except: raise RuntimeError("no input amount") if rst["type"] == "MULTISIG": return self.__sign_p2wsh_multisig(n, private_key, public_key, script_pub_key, redeem_script, sighash_type, amount) else: return self.__sign_p2wsh_custom(n, private_key, public_key, script_pub_key, redeem_script, sighash_type, amount) def __sign_p2wsh_multisig(self, n, private_key, public_key, script_pub_key, redeem_script, sighash_type, amount): script_code = int_to_var_int(len(redeem_script)) + redeem_script sighash = self.sig_hash_segwit(n, amount, script_pub_key=script_code, sighash_type=sighash_type) sighash = bytes.fromhex(sighash) if isinstance(sighash, str) else sighash sig = [sign_message(sighash, p, 0) + bytes([sighash_type]) for p in private_key] self["vIn"][n]['signatures'] = [s if self["format"] == "raw" else s.hex() for s in sig] if "txInWitness" not in self["vIn"][n]: self["vIn"][n]["txInWitness"] = [] witness = self.__get_multisig_script_sig__(self["vIn"][n]["txInWitness"], public_key, sig, script_code, redeem_script, n, amount) if self["format"] == "raw": self["vIn"][n]['txInWitness'] = list(witness) else: self["vIn"][n]["txInWitness"] = list([w.hex() for w in witness]) return b"" def __sign_p2wsh_custom(self, n, private_key, public_key, script_pub_key, redeem_script, sighash_type, amount): raise RuntimeError("not implemented __sign_p2wsh_custom") def __sign_p2sh_p2wsh_multisig(self, n, private_key, public_key, redeem_script, sighash_type, amount): self["segwit"] = True script_code = int_to_var_int(len(redeem_script)) + redeem_script sighash = self.sig_hash_segwit(n, amount, script_pub_key=script_code, sighash_type=sighash_type) sighash = bytes.fromhex(sighash) if isinstance(sighash, str) else sighash sig = [sign_message(sighash, p, 0) + bytes([sighash_type]) for p in private_key] self["vIn"][n]['signatures'] = [s if self["format"] == "raw" else s.hex() for s in sig] if "txInWitness" not in self["vIn"][n]: self["vIn"][n]["txInWitness"] = [] witness = self.__get_multisig_script_sig__(self["vIn"][n]["txInWitness"], public_key, sig, script_code, redeem_script, n, amount) if self["format"] == "raw": self["vIn"][n]['txInWitness'] = list(witness) else: self["vIn"][n]["txInWitness"] = list([w.hex() for w in witness]) # calculate P2SH redeem script from P2WSH redeem script return op_push_data(b"\x00" + op_push_data(sha256(redeem_script))) def __get_bare_multisig_script_sig__(self, script_sig, script_pub_key, keys, signatures, n): sig_map = {keys[i]:signatures[i] for i in range(len(keys))} pub_keys = get_multisig_public_keys(script_pub_key) s = get_stream(script_sig) o, d = read_opcode(s) while o: o, d = read_opcode(s) if d and is_valid_signature_encoding(d): for i in range(4): sighash = self.sig_hash(n, script_pub_key=script_pub_key, sighash_type=d[-1]) sighash = s2rh(sighash) if isinstance(sighash, str) else sighash pk = public_key_recovery(d[:-1], sighash, i, hex=0) if pk in pub_keys: sig_map[pk] = d break # recreate script sig r = [OP_0] for k in pub_keys: try: r.append(op_push_data(sig_map[k])) except: pass return r def __get_multisig_script_sig__(self, script_sig, keys, signatures, script_code, redeem_script, n, amount=None): sig_map = {keys[i]:signatures[i] for i in range(len(keys))} pub_keys = get_multisig_public_keys(redeem_script) p2wsh = True if isinstance(script_sig, list) else False if not p2wsh: s = get_stream(script_sig) o, d = read_opcode(s) while o: o, d = read_opcode(s) if d and is_valid_signature_encoding(d): for i in range(4): sighash = self.sig_hash(n, script_pub_key=script_code, sighash_type=d[-1]) sighash = s2rh(sighash) if isinstance(sighash, str) else sighash pk = public_key_recovery(d[:-1], sighash, i, hex=0) if pk in pub_keys: sig_map[pk] = d break # recreate script sig r = [OP_0] for k in pub_keys: try: r.append(op_push_data(sig_map[k])) except: pass r += [op_push_data(redeem_script)] else: for w in script_sig: if isinstance(w, str): w = bytes.fromhex(w) if w and is_valid_signature_encoding(w): d = w[:-1] for i in range(4): sighash = self.sig_hash_segwit(n, amount, script_pub_key=script_code, sighash_type=w[-1]) pk = public_key_recovery(d, sighash, i, hex=0) if pk in pub_keys: sig_map[pk] = w break r = [b""] for k in pub_keys: try: r.append(sig_map[k]) except: pass r += [redeem_script] return r def sig_hash(self, n, script_pub_key=None, sighash_type=SIGHASH_ALL, preimage=False): try: self["vIn"][n] except: raise Exception("sig_hash error, input not exist") # check script_pub_key for input if script_pub_key is not None: script_code = script_pub_key else: if "scriptPubKey" not in self["vIn"][n]: raise Exception("sig_hash error, scriptPubKey required") script_code = self["vIn"][n]["scriptPubKey"] if isinstance(script_code, str): script_code = bytes.fromhex(script_code) if not isinstance(script_code,bytes): raise Exception("sig_hash error, script_code type error") # remove opcode separators if ((sighash_type & 31) == SIGHASH_SINGLE) and (n >= (len(self["vOut"]))): if self["format"] == "raw": return b'\x01%s' % (b'\x00' * 31) return rh2s(b'\x01%s' % (b'\x00' * 31)) script_code = delete_from_script(script_code, BYTE_OPCODE["OP_CODESEPARATOR"]) pm = bytearray() pm += b"%s%s" % (pack(' n and (sighash_type & 31) == SIGHASH_SINGLE: continue if (sighash_type & 31) == SIGHASH_SINGLE and (n != i): pm += b"%s%s" % (b'\xff' * 8, b'\x00') else: pm += b"%s%s%s" % (self["vOut"][i]["value"].to_bytes(8, 'little'), int_to_var_int(len(script_pub_key)), script_pub_key) pm += b"%s%s" % (self["lockTime"].to_bytes(4, 'little'), pack(b"