from struct import unpack import json from .functions import * from .address import PrivateKey, Address, PublicKey, ScriptAddress 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): if format not in ("decoded", "raw"): raise ValueError("format error, raw or decoded allowed") 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() # 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.fromhex(w) if self["format"] == "raw" else w) else: witness.append(w if self["format"] == "raw" else bytes.fromhex(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.fromhex(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.fromhex(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) if script_pub_key: assert script_pub_key == 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 self.__refresh__() return self def add_output(self, amount, address=None, script_pub_key=None): assert address is not None or script_pub_key is not None assert not (address is None and script_pub_key is None) assert type(amount) == int assert amount >= 0 and amount <= MAX_AMOUNT if script_pub_key: if type(script_pub_key) == str: script_pub_key = bytes.fromhex(script_pub_key) assert type(script_pub_key) == bytes 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): 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): 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) self.__refresh__() 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 self.__refresh__() 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 self.__refresh__() 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) self.__refresh__() 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] 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]) 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]) 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] 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()] 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()] 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] 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] 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): # check n assert n >= 0 tx_in_count = len(self["vIn"]) if n >= tx_in_count: if self["format"] == "raw": return b'\x01' + b'\x00' * 31 else: return rh2s(b'\x01' + b'\x00' * 31) # check script_pub_key for input if script_pub_key is not None: script_code = script_pub_key else: assert "scriptPubKey" in self["vIn"][n] script_code = self["vIn"][n]["scriptPubKey"] if type(script_code) == str: script_code = bytes.fromhex(script_code) assert type(script_code) == bytes # remove opcode separators script_code = delete_from_script(script_code, BYTE_OPCODE["OP_CODESEPARATOR"]) pm = bytearray() if ((sighash_type & 31) == SIGHASH_SINGLE) and (n >= (len(self["vOut"]))): if self["format"] == "raw": return b'\x01' + b'\x00' * 31 else: return rh2s(b'\x01' + b'\x00' * 31) pm += struct.pack(' n and (sighash_type & 31) == SIGHASH_SINGLE: continue if (sighash_type & 31) == SIGHASH_SINGLE and (n != i): pm += b'\xff' * 8 + b'\x00' else: pm += self["vOut"][i]["value"].to_bytes(8, 'little') pm += int_to_var_int(len(script_pub_key)) + script_pub_key pm += self["lockTime"].to_bytes(4, 'little') pm += struct.pack(b"= 0 tx_in_count = len(self["vIn"]) if n >= tx_in_count: if self["format"] == "raw": return b'\x01' + b'\x00' * 31 else: return rh2s(b'\x01' + b'\x00' * 31) # check script_pub_key for input if script_pub_key is not None: script_code = script_pub_key else: assert "scriptPubKey" in self["vIn"][n] script_code = self["vIn"][n]["scriptPubKey"] if type(script_code) == str: script_code = bytes.fromhex(script_code) assert type(script_code) == bytes # remove opcode separators pm = bytearray() # 1. nVersion of the transaction (4-byte little endian) pm += struct.pack('