#!/usr/bin/env python

#

# Electrum - lightweight Bitcoin client

# Copyright (C) 2011 Thomas Voegtlin

#

# Permission is hereby granted, free of charge, to any person

# obtaining a copy of this software and associated documentation files

# (the "Software"), to deal in the Software without restriction,

# including without limitation the rights to use, copy, modify, merge,

# publish, distribute, sublicense, and/or sell copies of the Software,

# and to permit persons to whom the Software is furnished to do so,

# subject to the following conditions:

#

# The above copyright notice and this permission notice shall be

# included in all copies or substantial portions of the Software.

#

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,

# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF

# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND

# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS

# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN

# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN

# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE

# SOFTWARE.

# Note: The deserialization code originally comes from ABE.

import struct

import traceback

import sys

import io

import base64

from typing import (Sequence, Union, NamedTuple, Tuple, Optional, Iterable,

Callable, List, Dict, Set, TYPE_CHECKING)

from collections import defaultdict

from enum import IntEnum

import itertools

import binascii

from . import ecc, bitcoin, constants, segwit_addr, bip32

from .bip32 import BIP32Node

from .util import profiler, to_bytes, bh2u, bfh, chunks, is_hex_str

from .bitcoin import (TYPE_ADDRESS, TYPE_SCRIPT, hash_160,

hash160_to_p2sh, hash160_to_p2pkh, hash_to_segwit_addr,

var_int, TOTAL_COIN_SUPPLY_LIMIT_IN_BTC, COIN,

int_to_hex, push_script, b58_address_to_hash160,

opcodes, add_number_to_script, base_decode, is_segwit_script_type)

from .crypto import sha256d

from .logging import get_logger

if TYPE_CHECKING:

from .wallet import Abstract_Wallet

_logger = get_logger(__name__)

DEBUG_PSBT_PARSING = False

class SerializationError(Exception):

""" Thrown when there's a problem deserializing or serializing """

class UnknownTxinType(Exception):

pass

class BadHeaderMagic(SerializationError):

pass

class UnexpectedEndOfStream(SerializationError):

pass

class PSBTInputConsistencyFailure(SerializationError):

pass

class MalformedBitcoinScript(Exception):

pass

class MissingTxInputAmount(Exception):

pass

SIGHASH_ALL = 1

class TxOutput:

scriptpubkey: bytes

value: Union[int, str]

def __init__(self, *, scriptpubkey: bytes, value: Union[int, str]):

self.scriptpubkey = scriptpubkey

self.value = value # str when the output is set to max: '!' # in satoshis

@classmethod

def from_address_and_value(cls, address: str, value: Union[int, str]) -> Union['TxOutput', 'PartialTxOutput']:

return cls(scriptpubkey=bfh(bitcoin.address_to_script(address)),

value=value)

def serialize_to_network(self) -> bytes:

buf = int.to_bytes(self.value, 8, byteorder="little", signed=False)

script = self.scriptpubkey

buf += bfh(var_int(len(script.hex()) // 2))

buf += script

return buf

@classmethod

def from_network_bytes(cls, raw: bytes) -> 'TxOutput':

vds = BCDataStream()

vds.write(raw)

txout = parse_output(vds)

if vds.can_read_more():

raise SerializationError('extra junk at the end of TxOutput bytes')

return txout

def to_legacy_tuple(self) -> Tuple[int, str, Union[int, str]]:

if self.address:

return TYPE_ADDRESS, self.address, self.value

return TYPE_SCRIPT, self.scriptpubkey.hex(), self.value

@classmethod

def from_legacy_tuple(cls, _type: int, addr: str, val: Union[int, str]) -> Union['TxOutput', 'PartialTxOutput']:

if _type == TYPE_ADDRESS:

return cls.from_address_and_value(addr, val)

if _type == TYPE_SCRIPT:

return cls(scriptpubkey=bfh(addr), value=val)

raise Exception(f"unexptected legacy address type: {_type}")

@property

def address(self) -> Optional[str]:

return get_address_from_output_script(self.scriptpubkey) # TODO cache this?

def get_ui_address_str(self) -> str:

addr = self.address

if addr is not None:

return addr

return f"SCRIPT {self.scriptpubkey.hex()}"

def __repr__(self):

return f"<TxOutput script={self.scriptpubkey.hex()} address={self.address} value={self.value}>"

def __eq__(self, other):

if not isinstance(other, TxOutput):

return False

return self.scriptpubkey == other.scriptpubkey and self.value == other.value

def __ne__(self, other):

return not (self == other)

def to_json(self):

d = {

'scriptpubkey': self.scriptpubkey.hex(),

'address': self.address,

'value_sats': self.value,

}

return d

class BIP143SharedTxDigestFields(NamedTuple):

hashPrevouts: str

hashSequence: str

hashOutputs: str

class TxOutpoint(NamedTuple):

txid: bytes # endianness same as hex string displayed; reverse of tx serialization order

out_idx: int

@classmethod

def from_str(cls, s: str) -> 'TxOutpoint':

hash_str, idx_str = s.split(':')

return TxOutpoint(txid=bfh(hash_str),

out_idx=int(idx_str))

def to_str(self) -> str:

return f"{self.txid.hex()}:{self.out_idx}"

def serialize_to_network(self) -> bytes:

return self.txid[::-1] + bfh(int_to_hex(self.out_idx, 4))

def is_coinbase(self) -> bool:

return self.txid == bytes(32)

class TxInput:

prevout: TxOutpoint

script_sig: Optional[bytes]

nsequence: int

witness: Optional[bytes]

def __init__(self, *,

prevout: TxOutpoint,

script_sig: bytes = None,

nsequence: int = 0xffffffff - 1,

witness: bytes = None):

self.prevout = prevout

self.script_sig = script_sig

self.nsequence = nsequence

self.witness = witness

def is_coinbase(self) -> bool:

return self.prevout.is_coinbase()

def value_sats(self) -> Optional[int]:

return None

def to_json(self):

d = {

'prevout_hash': self.prevout.txid.hex(),

'prevout_n': self.prevout.out_idx,

'coinbase': self.is_coinbase(),

'nsequence': self.nsequence,

}

if self.script_sig is not None:

d['scriptSig'] = self.script_sig.hex()

if self.witness is not None:

d['witness'] = self.witness.hex()

return d

class BCDataStream(object):

"""Workalike python implementation of Bitcoin's CDataStream class."""

def __init__(self):

self.input = None # type: Optional[bytearray]

self.read_cursor = 0

def clear(self):

self.input = None

self.read_cursor = 0

def write(self, _bytes): # Initialize with string of _bytes

if self.input is None:

self.input = bytearray(_bytes)

else:

self.input += bytearray(_bytes)

def read_string(self, encoding='ascii'):

# Strings are encoded depending on length:

# 0 to 252 : 1-byte-length followed by bytes (if any)

# 253 to 65,535 : byte'253' 2-byte-length followed by bytes

# 65,536 to 4,294,967,295 : byte '254' 4-byte-length followed by bytes

# ... and the Bitcoin client is coded to understand:

# greater than 4,294,967,295 : byte '255' 8-byte-length followed by bytes of string

# ... but I don't think it actually handles any strings that big.

if self.input is None:

raise SerializationError("call write(bytes) before trying to deserialize")

length = self.read_compact_size()

return self.read_bytes(length).decode(encoding)

def write_string(self, string, encoding='ascii'):

string = to_bytes(string, encoding)

# Length-encoded as with read-string

self.write_compact_size(len(string))

self.write(string)

def read_bytes(self, length) -> bytes:

try:

result = self.input[self.read_cursor:self.read_cursor+length] # type: bytearray

self.read_cursor += length

return bytes(result)

except IndexError:

raise SerializationError("attempt to read past end of buffer") from None

def can_read_more(self) -> bool:

if not self.input:

return False

return self.read_cursor < len(self.input)

def read_boolean(self): return self.read_bytes(1)[0] != chr(0)

def read_int16(self): return self._read_num('<h')

def read_uint16(self): return self._read_num('<H')

def read_int32(self): return self._read_num('<i')

def read_uint32(self): return self._read_num('<I')

def read_int64(self): return self._read_num('<q')

def read_uint64(self): return self._read_num('<Q')

def write_boolean(self, val): return self.write(chr(1) if val else chr(0))

def write_int16(self, val): return self._write_num('<h', val)

def write_uint16(self, val): return self._write_num('<H', val)

def write_int32(self, val): return self._write_num('<i', val)

def write_uint32(self, val): return self._write_num('<I', val)

def write_int64(self, val): return self._write_num('<q', val)

def write_uint64(self, val): return self._write_num('<Q', val)

def read_compact_size(self):

try:

size = self.input[self.read_cursor]

self.read_cursor += 1

if size == 253:

size = self._read_num('<H')

elif size == 254:

size = self._read_num('<I')

elif size == 255:

size = self._read_num('<Q')

return size

except IndexError as e:

raise SerializationError("attempt to read past end of buffer") from e

def write_compact_size(self, size):

if size < 0:

raise SerializationError("attempt to write size < 0")

elif size < 253:

self.write(bytes([size]))

elif size < 2**16:

self.write(b'\xfd')

self._write_num('<H', size)

elif size < 2**32:

self.write(b'\xfe')

self._write_num('<I', size)

elif size < 2**64:

self.write(b'\xff')

self._write_num('<Q', size)

else:

raise Exception(f"size {size} too large for compact_size")

def _read_num(self, format):

try:

(i,) = struct.unpack_from(format, self.input, self.read_cursor)

self.read_cursor += struct.calcsize(format)

except Exception as e:

raise SerializationError(e) from e

return i

def _write_num(self, format, num):

s = struct.pack(format, num)

self.write(s)

def script_GetOp(_bytes : bytes):

i = 0

while i < len(_bytes):

vch = None

opcode = _bytes[i]

i += 1

if opcode <= opcodes.OP_PUSHDATA4:

nSize = opcode

if opcode == opcodes.OP_PUSHDATA1:

try: nSize = _bytes[i]

except IndexError: raise MalformedBitcoinScript()

i += 1

elif opcode == opcodes.OP_PUSHDATA2:

try: (nSize,) = struct.unpack_from('<H', _bytes, i)

except struct.error: raise MalformedBitcoinScript()

i += 2

elif opcode == opcodes.OP_PUSHDATA4:

try: (nSize,) = struct.unpack_from('<I', _bytes, i)

except struct.error: raise MalformedBitcoinScript()

i += 4

vch = _bytes[i:i + nSize]

i += nSize

yield opcode, vch, i

class OPPushDataGeneric:

def __init__(self, pushlen: Callable=None):

if pushlen is not None:

self.check_data_len = pushlen

@classmethod

def check_data_len(cls, datalen: int) -> bool:

# Opcodes below OP_PUSHDATA4 all just push data onto stack, and are equivalent.

return opcodes.OP_PUSHDATA4 >= datalen >= 0

@classmethod

def is_instance(cls, item):

# accept objects that are instances of this class

# or other classes that are subclasses

return isinstance(item, cls) \

or (isinstance(item, type) and issubclass(item, cls))

OPPushDataPubkey = OPPushDataGeneric(lambda x: x in (33, 65))

# note that this does not include x_pubkeys !

def match_decoded(decoded, to_match):

if decoded is None:

return False

if len(decoded) != len(to_match):

return False

for i in range(len(decoded)):

to_match_item = to_match[i]

decoded_item = decoded[i]

if OPPushDataGeneric.is_instance(to_match_item) and to_match_item.check_data_len(decoded_item[0]):

continue

if to_match_item != decoded_item[0]:

return False

return True

def get_address_from_output_script(_bytes: bytes, *, net=None) -> Optional[str]:

try:

decoded = [x for x in script_GetOp(_bytes)]

except MalformedBitcoinScript:

decoded = None

# p2pkh

match = [opcodes.OP_DUP, opcodes.OP_HASH160, OPPushDataGeneric(lambda x: x == 20), opcodes.OP_EQUALVERIFY, opcodes.OP_CHECKSIG]

if match_decoded(decoded, match):

return hash160_to_p2pkh(decoded[2][1], net=net)

# p2sh

match = [opcodes.OP_HASH160, OPPushDataGeneric(lambda x: x == 20), opcodes.OP_EQUAL]

if match_decoded(decoded, match):

return hash160_to_p2sh(decoded[1][1], net=net)

# segwit address (version 0)

match = [opcodes.OP_0, OPPushDataGeneric(lambda x: x in (20, 32))]

if match_decoded(decoded, match):

return hash_to_segwit_addr(decoded[1][1], witver=0, net=net)

# segwit address (version 1-16)

future_witness_versions = list(range(opcodes.OP_1, opcodes.OP_16 + 1))

for witver, opcode in enumerate(future_witness_versions, start=1):

match = [opcode, OPPushDataGeneric(lambda x: 2 <= x <= 40)]

if match_decoded(decoded, match):

return hash_to_segwit_addr(decoded[1][1], witver=witver, net=net)

return None

def parse_input(vds: BCDataStream) -> TxInput:

prevout_hash = vds.read_bytes(32)[::-1]

prevout_n = vds.read_uint32()

prevout = TxOutpoint(txid=prevout_hash, out_idx=prevout_n)

script_sig = vds.read_bytes(vds.read_compact_size())

nsequence = vds.read_uint32()

return TxInput(prevout=prevout, script_sig=script_sig, nsequence=nsequence)

def construct_witness(items: Sequence[Union[str, int, bytes]]) -> str:

"""Constructs a witness from the given stack items."""

witness = var_int(len(items))

for item in items:

if type(item) is int:

item = bitcoin.script_num_to_hex(item)

elif isinstance(item, (bytes, bytearray)):

item = bh2u(item)

witness += bitcoin.witness_push(item)

return witness

def parse_witness(vds: BCDataStream, txin: TxInput) -> None:

n = vds.read_compact_size()

witness_elements = list(vds.read_bytes(vds.read_compact_size()) for i in range(n))

txin.witness = bfh(construct_witness(witness_elements))

def parse_output(vds: BCDataStream) -> TxOutput:

value = vds.read_int64()

if value > TOTAL_COIN_SUPPLY_LIMIT_IN_BTC * COIN:

raise SerializationError('invalid output amount (too large)')

if value < 0:

raise SerializationError('invalid output amount (negative)')

scriptpubkey = vds.read_bytes(vds.read_compact_size())

return TxOutput(value=value, scriptpubkey=scriptpubkey)

# pay & redeem scripts

def multisig_script(public_keys: Sequence[str], m: int) -> str:

n = len(public_keys)

assert 1 <= m <= n <= 15, f'm {m}, n {n}'

op_m = bh2u(add_number_to_script(m))

op_n = bh2u(add_number_to_script(n))

keylist = [push_script(k) for k in public_keys]

return op_m + ''.join(keylist) + op_n + opcodes.OP_CHECKMULTISIG.hex()

class Transaction:

_cached_network_ser: Optional[str]

def __str__(self):

return self.serialize()

def __init__(self, raw):

if raw is None:

self._cached_network_ser = None

elif isinstance(raw, str):

self._cached_network_ser = raw.strip() if raw else None

assert is_hex_str(self._cached_network_ser)

elif isinstance(raw, (bytes, bytearray)):

self._cached_network_ser = bh2u(raw)

else:

raise Exception(f"cannot initialize transaction from {raw}")

self._inputs = None # type: List[TxInput]

self._outputs = None # type: List[TxOutput]

self.locktime = 0

self.version = 2

self._cached_txid = None # type: Optional[str]

def to_json(self) -> dict:

d = {

'version': self.version,

'locktime': self.locktime,

'inputs': [txin.to_json() for txin in self.inputs()],

'outputs': [txout.to_json() for txout in self.outputs()],

}

return d

def inputs(self) -> Sequence[TxInput]:

if self._inputs is None:

self.deserialize()

return self._inputs

def outputs(self) -> Sequence[TxOutput]:

if self._outputs is None:

self.deserialize()

return self._outputs

def deserialize(self) -> None:

if self._cached_network_ser is None:

return

if self._inputs is not None:

return

raw_bytes = bfh(self._cached_network_ser)

vds = BCDataStream()

vds.write(raw_bytes)

self.version = vds.read_int32()

n_vin = vds.read_compact_size()

is_segwit = (n_vin == 0)

if is_segwit:

marker = vds.read_bytes(1)

if marker != b'\x01':

raise ValueError('invalid txn marker byte: {}'.format(marker))

n_vin = vds.read_compact_size()

self._inputs = [parse_input(vds) for i in range(n_vin)]

n_vout = vds.read_compact_size()

self._outputs = [parse_output(vds) for i in range(n_vout)]

if is_segwit:

for txin in self._inputs:

parse_witness(vds, txin)

self.locktime = vds.read_uint32()

if vds.can_read_more():

raise SerializationError('extra junk at the end')

@classmethod

def get_siglist(self, txin: 'PartialTxInput', *, estimate_size=False):

if txin.prevout.is_coinbase():

return [], []

if estimate_size:

try:

pubkey_size = len(txin.pubkeys[0])

except IndexError:

pubkey_size = 33 # guess it is compressed

num_pubkeys = max(1, len(txin.pubkeys))

pk_list = ["00" * pubkey_size] * num_pubkeys

num_sig = max(1, txin.num_sig)

# we guess that signatures will be 72 bytes long

# note: DER-encoded ECDSA signatures are 71 or 72 bytes in practice

# See https://bitcoin.stackexchange.com/questions/77191/what-is-the-maximum-size-of-a-der-encoded-ecdsa-signature

# We assume low S (as that is a bitcoin standardness rule).

# We do not assume low R (even though the sigs we create conform), as external sigs,

# e.g. from a hw signer cannot be expected to have a low R.

sig_list = [ "00" * 72 ] * num_sig

else:

pk_list = [pubkey.hex() for pubkey in txin.pubkeys]

sig_list = [txin.part_sigs.get(pubkey, b'').hex() for pubkey in txin.pubkeys]

if txin.is_complete():

sig_list = [sig for sig in sig_list if sig]

return pk_list, sig_list

@classmethod

def serialize_witness(cls, txin: TxInput, *, estimate_size=False) -> str:

if txin.witness is not None:

return txin.witness.hex()

if txin.prevout.is_coinbase():

return ''

assert isinstance(txin, PartialTxInput)

_type = txin.script_type

if not cls.is_segwit_input(txin):

return '00'

if _type in ('address', 'unknown') and estimate_size:

_type = cls.guess_txintype_from_address(txin.address)

pubkeys, sig_list = cls.get_siglist(txin, estimate_size=estimate_size)

if _type in ['p2wpkh', 'p2wpkh-p2sh']:

return construct_witness([sig_list[0], pubkeys[0]])

elif _type in ['p2wsh', 'p2wsh-p2sh']:

witness_script = multisig_script(pubkeys, txin.num_sig)

return construct_witness([0] + sig_list + [witness_script])

elif _type in ['p2pk', 'p2pkh', 'p2sh']:

return '00'

raise UnknownTxinType(f'cannot construct witness for txin_type: {_type}')

@classmethod

def is_segwit_input(cls, txin: 'TxInput', *, guess_for_address=False) -> bool:

if txin.witness not in (b'\x00', b'', None):

return True

if not isinstance(txin, PartialTxInput):

return False

if txin.is_native_segwit() or txin.is_p2sh_segwit():

return True

if txin.is_native_segwit() is False and txin.is_p2sh_segwit() is False:

return False

if txin.witness_script:

return True

_type = txin.script_type

if _type == 'address' and guess_for_address:

_type = cls.guess_txintype_from_address(txin.address)

return is_segwit_script_type(_type)

@classmethod

def guess_txintype_from_address(cls, addr: Optional[str]) -> str:

# It's not possible to tell the script type in general

# just from an address.

# - "1" addresses are of course p2pkh

# - "3" addresses are p2sh but we don't know the redeem script..

# - "bc1" addresses (if they are 42-long) are p2wpkh

# - "bc1" addresses that are 62-long are p2wsh but we don't know the script..

# If we don't know the script, we _guess_ it is pubkeyhash.

# As this method is used e.g. for tx size estimation,

# the estimation will not be precise.

if addr is None:

return 'p2wpkh'

witver, witprog = segwit_addr.decode(constants.net.SEGWIT_HRP, addr)

if witprog is not None:

return 'p2wpkh'

addrtype, hash_160_ = b58_address_to_hash160(addr)

if addrtype == constants.net.ADDRTYPE_P2PKH:

return 'p2pkh'

elif addrtype == constants.net.ADDRTYPE_P2SH:

return 'p2wpkh-p2sh'

raise Exception(f'unrecognized address: {repr(addr)}')

@classmethod

def input_script(self, txin: TxInput, *, estimate_size=False) -> str:

if txin.script_sig is not None:

return txin.script_sig.hex()

if txin.prevout.is_coinbase():

return ''

assert isinstance(txin, PartialTxInput)

if txin.is_p2sh_segwit() and txin.redeem_script:

return push_script(txin.redeem_script.hex())

if txin.is_native_segwit():

return ''

_type = txin.script_type

pubkeys, sig_list = self.get_siglist(txin, estimate_size=estimate_size)

script = ''.join(push_script(x) for x in sig_list)

if _type in ('address', 'unknown') and estimate_size:

_type = self.guess_txintype_from_address(txin.address)

if _type == 'p2pk':

return script

elif _type == 'p2sh':

# put op_0 before script

script = '00' + script

redeem_script = multisig_script(pubkeys, txin.num_sig)

script += push_script(redeem_script)

return script

elif _type == 'p2pkh':

script += push_script(pubkeys[0])

return script

elif _type in ['p2wpkh', 'p2wsh']:

return ''

elif _type == 'p2wpkh-p2sh':

redeem_script = bitcoin.p2wpkh_nested_script(pubkeys[0])

return push_script(redeem_script)

elif _type == 'p2wsh-p2sh':

if estimate_size:

witness_script = ''

else:

witness_script = self.get_preimage_script(txin)

redeem_script = bitcoin.p2wsh_nested_script(witness_script)

return push_script(redeem_script)

raise UnknownTxinType(f'cannot construct scriptSig for txin_type: {_type}')

@classmethod

def get_preimage_script(cls, txin: 'PartialTxInput') -> str:

if txin.witness_script:

opcodes_in_witness_script = [x[0] for x in script_GetOp(txin.witness_script)]

if opcodes.OP_CODESEPARATOR in opcodes_in_witness_script:

raise Exception('OP_CODESEPARATOR black magic is not supported')

return txin.witness_script.hex()

pubkeys = [pk.hex() for pk in txin.pubkeys]

if txin.script_type in ['p2sh', 'p2wsh', 'p2wsh-p2sh']:

return multisig_script(pubkeys, txin.num_sig)

elif txin.script_type in ['p2pkh', 'p2wpkh', 'p2wpkh-p2sh']:

pubkey = pubkeys[0]

pkh = bh2u(hash_160(bfh(pubkey)))

return bitcoin.pubkeyhash_to_p2pkh_script(pkh)

elif txin.script_type == 'p2pk':

pubkey = pubkeys[0]

return bitcoin.public_key_to_p2pk_script(pubkey)

else:

raise UnknownTxinType(f'cannot construct preimage_script for txin_type: {txin.script_type}')

@classmethod

def serialize_input(self, txin: TxInput, script: str) -> str:

# Prev hash and index

s = txin.prevout.serialize_to_network().hex()

# Script length, script, sequence

s += var_int(len(script)//2)

s += script

s += int_to_hex(txin.nsequence, 4)

return s

def _calc_bip143_shared_txdigest_fields(self) -> BIP143SharedTxDigestFields:

inputs = self.inputs()

outputs = self.outputs()

hashPrevouts = bh2u(sha256d(b''.join(txin.prevout.serialize_to_network() for txin in inputs)))

hashSequence = bh2u(sha256d(bfh(''.join(int_to_hex(txin.nsequence, 4) for txin in inputs))))

hashOutputs = bh2u(sha256d(bfh(''.join(o.serialize_to_network().hex() for o in outputs))))

return BIP143SharedTxDigestFields(hashPrevouts=hashPrevouts,

hashSequence=hashSequence,

hashOutputs=hashOutputs)

def is_segwit(self, *, guess_for_address=False):

return any(self.is_segwit_input(txin, guess_for_address=guess_for_address)

for txin in self.inputs())

def invalidate_ser_cache(self):

self._cached_network_ser = None

self._cached_txid = None

def serialize(self) -> str:

if not self._cached_network_ser:

self._cached_network_ser = self.serialize_to_network(estimate_size=False, include_sigs=True)

return self._cached_network_ser

def serialize_as_bytes(self) -> bytes:

return bfh(self.serialize())

def serialize_to_network(self, *, estimate_size=False, include_sigs=True, force_legacy=False) -> str:

"""Serialize the transaction as used on the Bitcoin network, into hex.

`include_sigs` signals whether to include scriptSigs and witnesses.

`force_legacy` signals to use the pre-segwit format

note: (not include_sigs) implies force_legacy

"""

self.deserialize()

nVersion = int_to_hex(self.version, 4)

nLocktime = int_to_hex(self.locktime, 4)

inputs = self.inputs()

outputs = self.outputs()

def create_script_sig(txin: TxInput) -> str:

if include_sigs:

return self.input_script(txin, estimate_size=estimate_size)

return ''

txins = var_int(len(inputs)) + ''.join(self.serialize_input(txin, create_script_sig(txin))

for txin in inputs)

txouts = var_int(len(outputs)) + ''.join(o.serialize_to_network().hex() for o in outputs)

use_segwit_ser_for_estimate_size = estimate_size and self.is_segwit(guess_for_address=True)

use_segwit_ser_for_actual_use = not estimate_size and self.is_segwit()

use_segwit_ser = use_segwit_ser_for_estimate_size or use_segwit_ser_for_actual_use

if include_sigs and not force_legacy and use_segwit_ser:

marker = '00'

flag = '01'

witness = ''.join(self.serialize_witness(x, estimate_size=estimate_size) for x in inputs)

return nVersion + marker + flag + txins + txouts + witness + nLocktime

else:

return nVersion + txins + txouts + nLocktime

def txid(self) -> Optional[str]:

if self._cached_txid is None:

self.deserialize()

all_segwit = all(self.is_segwit_input(x) for x in self.inputs())

if not all_segwit and not self.is_complete():

return None

try:

ser = self.serialize_to_network(force_legacy=True)

except UnknownTxinType:

# we might not know how to construct scriptSig for some scripts

return None

self._cached_txid = bh2u(sha256d(bfh(ser))[::-1])

return self._cached_txid

def wtxid(self) -> Optional[str]:

self.deserialize()

if not self.is_complete():

return None

try:

ser = self.serialize_to_network()

except UnknownTxinType:

# we might not know how to construct scriptSig/witness for some scripts

return None

return bh2u(sha256d(bfh(ser))[::-1])

def add_info_from_wallet(self, wallet: 'Abstract_Wallet') -> None:

return # no-op

def is_final(self):

return not any([txin.nsequence < 0xffffffff - 1 for txin in self.inputs()])

def estimated_size(self):

"""Return an estimated virtual tx size in vbytes.

BIP-0141 defines 'Virtual transaction size' to be weight/4 rounded up.

This definition is only for humans, and has little meaning otherwise.

If we wanted sub-byte precision, fee calculation should use transaction

weights, but for simplicity we approximate that with (virtual_size)x4

"""

weight = self.estimated_weight()

return self.virtual_size_from_weight(weight)

@classmethod

def estimated_input_weight(cls, txin, is_segwit_tx):

'''Return an estimate of serialized input weight in weight units.'''

script = cls.input_script(txin, estimate_size=True)

input_size = len(cls.serialize_input(txin, script)) // 2

if cls.is_segwit_input(txin, guess_for_address=True):

witness_size = len(cls.serialize_witness(txin, estimate_size=True)) // 2

else:

witness_size = 1 if is_segwit_tx else 0

return 4 * input_size + witness_size

@classmethod

def estimated_output_size(cls, address):

"""Return an estimate of serialized output size in bytes."""

script = bitcoin.address_to_script(address)

# 8 byte value + 1 byte script len + script

return 9 + len(script) // 2

@classmethod

def virtual_size_from_weight(cls, weight):

return weight // 4 + (weight % 4 > 0)

def estimated_total_size(self):

"""Return an estimated total transaction size in bytes."""

if not self.is_complete() or self._cached_network_ser is None:

return len(self.serialize_to_network(estimate_size=True)) // 2

else:

return len(self._cached_network_ser) // 2 # ASCII hex string

def estimated_witness_size(self):

"""Return an estimate of witness size in bytes."""

estimate = not self.is_complete()

if not self.is_segwit(guess_for_address=estimate):

return 0

inputs = self.inputs()

witness = ''.join(self.serialize_witness(x, estimate_size=estimate) for x in inputs)

witness_size = len(witness) // 2 + 2 # include marker and flag

return witness_size

def estimated_base_size(self):

"""Return an estimated base transaction size in bytes."""

return self.estimated_total_size() - self.estimated_witness_size()

def estimated_weight(self):

"""Return an estimate of transaction weight."""

total_tx_size = self.estimated_total_size()

base_tx_size = self.estimated_base_size()

return 3 * base_tx_size + total_tx_size

def is_complete(self) -> bool:

return True

def get_output_idxs_from_scriptpubkey(self, script: str) -> Set[int]:

"""Returns the set indices of outputs with given script."""

assert isinstance(script, str) # hex

# build cache if there isn't one yet

# note: can become stale and return incorrect data

# if the tx is modified later; that's out of scope.

if not hasattr(self, '_script_to_output_idx'):

d = defaultdict(set)

for output_idx, o in enumerate(self.outputs()):

o_script = o.scriptpubkey.hex()

assert isinstance(o_script, str)

d[o_script].add(output_idx)

self._script_to_output_idx = d

return set(self._script_to_output_idx[script]) # copy

def get_output_idxs_from_address(self, addr: str) -> Set[int]:

script = bitcoin.address_to_script(addr)

return self.get_output_idxs_from_scriptpubkey(script)

def output_value_for_address(self, addr):

# assumes exactly one output has that address

for o in self.outputs():

if o.address == addr:

return o.value

else:

raise Exception('output not found', addr)

def convert_raw_tx_to_hex(raw: Union[str, bytes]) -> str:

"""Sanitizes tx-describing input (hex/base43/base64) into

raw tx hex string."""

if not raw:

raise ValueError("empty string")

raw_unstripped = raw

raw = raw.strip()

# try hex

try:

return binascii.unhexlify(raw).hex()

except:

pass

# try base43

try:

return base_decode(raw, base=43).hex()

except:

pass

# try base64

if raw[0:6] in ('cHNidP', b'cHNidP'): # base64 psbt

try:

return base64.b64decode(raw).hex()

except:

pass

# raw bytes (do not strip whitespaces in this case)

if isinstance(raw_unstripped, bytes):

return raw_unstripped.hex()

raise ValueError(f"failed to recognize transaction encoding for txt: {raw[:30]}...")

def tx_from_any(raw: Union[str, bytes]) -> Union['PartialTransaction', 'Transaction']:

if isinstance(raw, bytearray):

raw = bytes(raw)

raw = convert_raw_tx_to_hex(raw)

try:

return PartialTransaction.from_raw_psbt(raw)

except BadHeaderMagic:

if raw[:10] == b'EPTF\xff'.hex():

raise SerializationError("Partial transactions generated with old Electrum versions "

"(< 4.0) are no longer supported. Please upgrade Electrum on "

"the other machine where this transaction was created.")

try:

tx = Transaction(raw)

tx.deserialize()

return tx

except Exception as e:

raise SerializationError(f"Failed to recognise tx encoding, or to parse transaction. "

f"raw: {raw[:30]}...") from e

class PSBTGlobalType(IntEnum):

UNSIGNED_TX = 0

XPUB = 1

VERSION = 0xFB

class PSBTInputType(IntEnum):

NON_WITNESS_UTXO = 0

WITNESS_UTXO = 1

PARTIAL_SIG = 2

SIGHASH_TYPE = 3

REDEEM_SCRIPT = 4

WITNESS_SCRIPT = 5

BIP32_DERIVATION = 6

FINAL_SCRIPTSIG = 7

FINAL_SCRIPTWITNESS = 8

class PSBTOutputType(IntEnum):

REDEEM_SCRIPT = 0

WITNESS_SCRIPT = 1

BIP32_DERIVATION = 2

# Serialization/deserialization tools

def deser_compact_size(f) -> Optional[int]:

try:

nit = f.read(1)[0]

except IndexError:

return None # end of file

if nit == 253:

nit = struct.unpack("<H", f.read(2))[0]

elif nit == 254:

nit = struct.unpack("<I", f.read(4))[0]

elif nit == 255:

nit = struct.unpack("<Q", f.read(8))[0]

return nit

class PSBTSection:

def _populate_psbt_fields_from_fd(self, fd=None):

if not fd: return

while True:

try:

key_type, key, val = self.get_next_kv_from_fd(fd)

except StopIteration:

break

self.parse_psbt_section_kv(key_type, key, val)

@classmethod

def get_next_kv_from_fd(cls, fd) -> Tuple[int, bytes, bytes]:

key_size = deser_compact_size(fd)

if key_size == 0: